ceph/src/rocksdb/db/repair.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 // Repairer does best effort recovery to recover as much data as possible after
  11 // a disaster without compromising consistency. It does not guarantee bringing
  12 // the database to a time consistent state.
  13 //
  14 // Repair process is broken into 4 phases:
  15 // (a) Find files
  16 // (b) Convert logs to tables
  17 // (c) Extract metadata
  18 // (d) Write Descriptor
  19 //
  20 // (a) Find files
  21 //
  22 // The repairer goes through all the files in the directory, and classifies them
  23 // based on their file name. Any file that cannot be identified by name will be
  24 // ignored.
  25 //
  26 // (b) Convert logs to table
  27 //
  28 // Every log file that is active is replayed. All sections of the file where the
  29 // checksum does not match is skipped over. We intentionally give preference to
  30 // data consistency.
  31 //
  32 // (c) Extract metadata
  33 //
  34 // We scan every table to compute
  35 // (1) smallest/largest for the table
  36 // (2) largest sequence number in the table
  37 // (3) oldest blob file referred to by the table (if applicable)
  38 //
  39 // If we are unable to scan the file, then we ignore the table.
  40 //
  41 // (d) Write Descriptor
  42 //
  43 // We generate descriptor contents:
  44 //  - log number is set to zero
  45 //  - next-file-number is set to 1 + largest file number we found
  46 //  - last-sequence-number is set to largest sequence# found across
  47 //    all tables (see 2c)
  48 //  - compaction pointers are cleared
  49 //  - every table file is added at level 0
  50 //
  51 // Possible optimization 1:
  52 //   (a) Compute total size and use to pick appropriate max-level M
  53 //   (b) Sort tables by largest sequence# in the table
  54 //   (c) For each table: if it overlaps earlier table, place in level-0,
  55 //       else place in level-M.
  56 //   (d) We can provide options for time consistent recovery and unsafe recovery
  57 //       (ignore checksum failure when applicable)
  58 // Possible optimization 2:
  59 //   Store per-table metadata (smallest, largest, largest-seq#, ...)
  60 //   in the table's meta section to speed up ScanTable.
  61
  62 #ifndef ROCKSDB_LITE
  63
  64 #include <cinttypes>
  65 #include "db/builder.h"
  66 #include "db/db_impl/db_impl.h"
  67 #include "db/dbformat.h"
  68 #include "db/log_reader.h"
  69 #include "db/log_writer.h"
  70 #include "db/memtable.h"
  71 #include "db/table_cache.h"
  72 #include "db/version_edit.h"
  73 #include "db/write_batch_internal.h"
  74 #include "env/composite_env_wrapper.h"
  75 #include "file/filename.h"
  76 #include "file/writable_file_writer.h"
  77 #include "options/cf_options.h"
  78 #include "rocksdb/comparator.h"
  79 #include "rocksdb/db.h"
  80 #include "rocksdb/env.h"
  81 #include "rocksdb/options.h"
  82 #include "rocksdb/write_buffer_manager.h"
  83 #include "table/scoped_arena_iterator.h"
  84 #include "util/string_util.h"
  85
  86 namespace ROCKSDB_NAMESPACE {
  87
  88 namespace {
  89
  90 class Repairer {
  91  public:
  92   Repairer(const std::string& dbname, const DBOptions& db_options,
  93            const std::vector<ColumnFamilyDescriptor>& column_families,
  94            const ColumnFamilyOptions& default_cf_opts,
  95            const ColumnFamilyOptions& unknown_cf_opts, bool create_unknown_cfs)
  96       : dbname_(dbname),
  97         env_(db_options.env),
  98         env_options_(),
  99         db_options_(SanitizeOptions(dbname_, db_options)),
 100         immutable_db_options_(ImmutableDBOptions(db_options_)),
 101         icmp_(default_cf_opts.comparator),
 102         default_cf_opts_(
 103             SanitizeOptions(immutable_db_options_, default_cf_opts)),
 104         default_cf_iopts_(
 105             ImmutableCFOptions(immutable_db_options_, default_cf_opts_)),
 106         unknown_cf_opts_(
 107             SanitizeOptions(immutable_db_options_, unknown_cf_opts)),
 108         create_unknown_cfs_(create_unknown_cfs),
 109         raw_table_cache_(
 110             // TableCache can be small since we expect each table to be opened
 111             // once.
 112             NewLRUCache(10, db_options_.table_cache_numshardbits)),
 113         table_cache_(new TableCache(default_cf_iopts_, env_options_,
 114                                     raw_table_cache_.get(),
 115                                     /*block_cache_tracer=*/nullptr)),
 116         wb_(db_options_.db_write_buffer_size),
 117         wc_(db_options_.delayed_write_rate),
 118         vset_(dbname_, &immutable_db_options_, env_options_,
 119               raw_table_cache_.get(), &wb_, &wc_,
 120               /*block_cache_tracer=*/nullptr),
 121         next_file_number_(1),
 122         db_lock_(nullptr) {
 123     for (const auto& cfd : column_families) {
 124       cf_name_to_opts_[cfd.name] = cfd.options;
 125     }
 126   }
 127
 128   const ColumnFamilyOptions* GetColumnFamilyOptions(
 129       const std::string& cf_name) {
 130     if (cf_name_to_opts_.find(cf_name) == cf_name_to_opts_.end()) {
 131       if (create_unknown_cfs_) {
 132         return &unknown_cf_opts_;
 133       }
 134       return nullptr;
 135     }
 136     return &cf_name_to_opts_[cf_name];
 137   }
 138
 139   // Adds a column family to the VersionSet with cf_options_ and updates
 140   // manifest.
 141   Status AddColumnFamily(const std::string& cf_name, uint32_t cf_id) {
 142     const auto* cf_opts = GetColumnFamilyOptions(cf_name);
 143     if (cf_opts == nullptr) {
 144       return Status::Corruption("Encountered unknown column family with name=" +
 145                                 cf_name + ", id=" + ToString(cf_id));
 146     }
 147     Options opts(db_options_, *cf_opts);
 148     MutableCFOptions mut_cf_opts(opts);
 149
 150     VersionEdit edit;
 151     edit.SetComparatorName(opts.comparator->Name());
 152     edit.SetLogNumber(0);
 153     edit.SetColumnFamily(cf_id);
 154     ColumnFamilyData* cfd;
 155     cfd = nullptr;
 156     edit.AddColumnFamily(cf_name);
 157
 158     mutex_.Lock();
 159     Status status = vset_.LogAndApply(cfd, mut_cf_opts, &edit, &mutex_,
 160                                       nullptr /* db_directory */,
 161                                       false /* new_descriptor_log */, cf_opts);
 162     mutex_.Unlock();
 163     return status;
 164   }
 165
 166   ~Repairer() {
 167     if (db_lock_ != nullptr) {
 168       env_->UnlockFile(db_lock_);
 169     }
 170     delete table_cache_;
 171   }
 172
 173   Status Run() {
 174     Status status = env_->LockFile(LockFileName(dbname_), &db_lock_);
 175     if (!status.ok()) {
 176       return status;
 177     }
 178     status = FindFiles();
 179     if (status.ok()) {
 180       // Discard older manifests and start a fresh one
 181       for (size_t i = 0; i < manifests_.size(); i++) {
 182         ArchiveFile(dbname_ + "/" + manifests_[i]);
 183       }
 184       // Just create a DBImpl temporarily so we can reuse NewDB()
 185       DBImpl* db_impl = new DBImpl(db_options_, dbname_);
 186       status = db_impl->NewDB();
 187       delete db_impl;
 188     }
 189
 190     if (status.ok()) {
 191       // Recover using the fresh manifest created by NewDB()
 192       status =
 193           vset_.Recover({{kDefaultColumnFamilyName, default_cf_opts_}}, false);
 194     }
 195     if (status.ok()) {
 196       // Need to scan existing SST files first so the column families are
 197       // created before we process WAL files
 198       ExtractMetaData();
 199
 200       // ExtractMetaData() uses table_fds_ to know which SST files' metadata to
 201       // extract -- we need to clear it here since metadata for existing SST
 202       // files has been extracted already
 203       table_fds_.clear();
 204       ConvertLogFilesToTables();
 205       ExtractMetaData();
 206       status = AddTables();
 207     }
 208     if (status.ok()) {
 209       uint64_t bytes = 0;
 210       for (size_t i = 0; i < tables_.size(); i++) {
 211         bytes += tables_[i].meta.fd.GetFileSize();
 212       }
 213       ROCKS_LOG_WARN(db_options_.info_log,
 214                      "**** Repaired rocksdb %s; "
 215                      "recovered %" ROCKSDB_PRIszt " files; %" PRIu64
 216                      " bytes. "
 217                      "Some data may have been lost. "
 218                      "****",
 219                      dbname_.c_str(), tables_.size(), bytes);
 220     }
 221     return status;
 222   }
 223
 224  private:
 225   struct TableInfo {
 226     FileMetaData meta;
 227     uint32_t column_family_id;
 228     std::string column_family_name;
 229   };
 230
 231   std::string const dbname_;
 232   Env* const env_;
 233   const EnvOptions env_options_;
 234   const DBOptions db_options_;
 235   const ImmutableDBOptions immutable_db_options_;
 236   const InternalKeyComparator icmp_;
 237   const ColumnFamilyOptions default_cf_opts_;
 238   const ImmutableCFOptions default_cf_iopts_;  // table_cache_ holds reference
 239   const ColumnFamilyOptions unknown_cf_opts_;
 240   const bool create_unknown_cfs_;
 241   std::shared_ptr<Cache> raw_table_cache_;
 242   TableCache* table_cache_;
 243   WriteBufferManager wb_;
 244   WriteController wc_;
 245   VersionSet vset_;
 246   std::unordered_map<std::string, ColumnFamilyOptions> cf_name_to_opts_;
 247   InstrumentedMutex mutex_;
 248
 249   std::vector<std::string> manifests_;
 250   std::vector<FileDescriptor> table_fds_;
 251   std::vector<uint64_t> logs_;
 252   std::vector<TableInfo> tables_;
 253   uint64_t next_file_number_;
 254   // Lock over the persistent DB state. Non-nullptr iff successfully
 255   // acquired.
 256   FileLock* db_lock_;
 257
 258   Status FindFiles() {
 259     std::vector<std::string> filenames;
 260     bool found_file = false;
 261     std::vector<std::string> to_search_paths;
 262
 263     for (size_t path_id = 0; path_id < db_options_.db_paths.size(); path_id++) {
 264         to_search_paths.push_back(db_options_.db_paths[path_id].path);
 265     }
 266
 267     // search wal_dir if user uses a customize wal_dir
 268     bool same = false;
 269     Status status = env_->AreFilesSame(db_options_.wal_dir, dbname_, &same);
 270     if (status.IsNotSupported()) {
 271       same = db_options_.wal_dir == dbname_;
 272       status = Status::OK();
 273     } else if (!status.ok()) {
 274       return status;
 275     }
 276
 277     if (!same) {
 278       to_search_paths.push_back(db_options_.wal_dir);
 279     }
 280
 281     for (size_t path_id = 0; path_id < to_search_paths.size(); path_id++) {
 282       status = env_->GetChildren(to_search_paths[path_id], &filenames);
 283       if (!status.ok()) {
 284         return status;
 285       }
 286       if (!filenames.empty()) {
 287         found_file = true;
 288       }
 289
 290       uint64_t number;
 291       FileType type;
 292       for (size_t i = 0; i < filenames.size(); i++) {
 293         if (ParseFileName(filenames[i], &number, &type)) {
 294           if (type == kDescriptorFile) {
 295             manifests_.push_back(filenames[i]);
 296           } else {
 297             if (number + 1 > next_file_number_) {
 298               next_file_number_ = number + 1;
 299             }
 300             if (type == kLogFile) {
 301               logs_.push_back(number);
 302             } else if (type == kTableFile) {
 303               table_fds_.emplace_back(number, static_cast<uint32_t>(path_id),
 304                                       0);
 305             } else {
 306               // Ignore other files
 307             }
 308           }
 309         }
 310       }
 311     }
 312     if (!found_file) {
 313       return Status::Corruption(dbname_, "repair found no files");
 314     }
 315     return Status::OK();
 316   }
 317
 318   void ConvertLogFilesToTables() {
 319     for (size_t i = 0; i < logs_.size(); i++) {
 320       // we should use LogFileName(wal_dir, logs_[i]) here. user might uses wal_dir option.
 321       std::string logname = LogFileName(db_options_.wal_dir, logs_[i]);
 322       Status status = ConvertLogToTable(logs_[i]);
 323       if (!status.ok()) {
 324         ROCKS_LOG_WARN(db_options_.info_log,
 325                        "Log #%" PRIu64 ": ignoring conversion error: %s",
 326                        logs_[i], status.ToString().c_str());
 327       }
 328       ArchiveFile(logname);
 329     }
 330   }
 331
 332   Status ConvertLogToTable(uint64_t log) {
 333     struct LogReporter : public log::Reader::Reporter {
 334       Env* env;
 335       std::shared_ptr<Logger> info_log;
 336       uint64_t lognum;
 337       void Corruption(size_t bytes, const Status& s) override {
 338         // We print error messages for corruption, but continue repairing.
 339         ROCKS_LOG_ERROR(info_log, "Log #%" PRIu64 ": dropping %d bytes; %s",
 340                         lognum, static_cast<int>(bytes), s.ToString().c_str());
 341       }
 342     };
 343
 344     // Open the log file
 345     std::string logname = LogFileName(db_options_.wal_dir, log);
 346     std::unique_ptr<SequentialFile> lfile;
 347     Status status = env_->NewSequentialFile(
 348         logname, &lfile, env_->OptimizeForLogRead(env_options_));
 349     if (!status.ok()) {
 350       return status;
 351     }
 352     std::unique_ptr<SequentialFileReader> lfile_reader(new SequentialFileReader(
 353         NewLegacySequentialFileWrapper(lfile), logname));
 354
 355     // Create the log reader.
 356     LogReporter reporter;
 357     reporter.env = env_;
 358     reporter.info_log = db_options_.info_log;
 359     reporter.lognum = log;
 360     // We intentionally make log::Reader do checksumming so that
 361     // corruptions cause entire commits to be skipped instead of
 362     // propagating bad information (like overly large sequence
 363     // numbers).
 364     log::Reader reader(db_options_.info_log, std::move(lfile_reader), &reporter,
 365                        true /*enable checksum*/, log);
 366
 367     // Initialize per-column family memtables
 368     for (auto* cfd : *vset_.GetColumnFamilySet()) {
 369       cfd->CreateNewMemtable(*cfd->GetLatestMutableCFOptions(),
 370                              kMaxSequenceNumber);
 371     }
 372     auto cf_mems = new ColumnFamilyMemTablesImpl(vset_.GetColumnFamilySet());
 373
 374     // Read all the records and add to a memtable
 375     std::string scratch;
 376     Slice record;
 377     WriteBatch batch;
 378     int counter = 0;
 379     while (reader.ReadRecord(&record, &scratch)) {
 380       if (record.size() < WriteBatchInternal::kHeader) {
 381         reporter.Corruption(
 382             record.size(), Status::Corruption("log record too small"));
 383         continue;
 384       }
 385       WriteBatchInternal::SetContents(&batch, record);
 386       status =
 387           WriteBatchInternal::InsertInto(&batch, cf_mems, nullptr, nullptr);
 388       if (status.ok()) {
 389         counter += WriteBatchInternal::Count(&batch);
 390       } else {
 391         ROCKS_LOG_WARN(db_options_.info_log, "Log #%" PRIu64 ": ignoring %s",
 392                        log, status.ToString().c_str());
 393         status = Status::OK();  // Keep going with rest of file
 394       }
 395     }
 396
 397     // Dump a table for each column family with entries in this log file.
 398     for (auto* cfd : *vset_.GetColumnFamilySet()) {
 399       // Do not record a version edit for this conversion to a Table
 400       // since ExtractMetaData() will also generate edits.
 401       MemTable* mem = cfd->mem();
 402       if (mem->IsEmpty()) {
 403         continue;
 404       }
 405
 406       FileMetaData meta;
 407       meta.fd = FileDescriptor(next_file_number_++, 0, 0);
 408       ReadOptions ro;
 409       ro.total_order_seek = true;
 410       Arena arena;
 411       ScopedArenaIterator iter(mem->NewIterator(ro, &arena));
 412       int64_t _current_time = 0;
 413       status = env_->GetCurrentTime(&_current_time);  // ignore error
 414       const uint64_t current_time = static_cast<uint64_t>(_current_time);
 415       SnapshotChecker* snapshot_checker = DisableGCSnapshotChecker::Instance();
 416
 417       auto write_hint = cfd->CalculateSSTWriteHint(0);
 418       std::vector<std::unique_ptr<FragmentedRangeTombstoneIterator>>
 419           range_del_iters;
 420       auto range_del_iter =
 421           mem->NewRangeTombstoneIterator(ro, kMaxSequenceNumber);
 422       if (range_del_iter != nullptr) {
 423         range_del_iters.emplace_back(range_del_iter);
 424       }
 425
 426       LegacyFileSystemWrapper fs(env_);
 427       status = BuildTable(
 428           dbname_, env_, &fs, *cfd->ioptions(),
 429           *cfd->GetLatestMutableCFOptions(), env_options_, table_cache_,
 430           iter.get(), std::move(range_del_iters), &meta,
 431           cfd->internal_comparator(), cfd->int_tbl_prop_collector_factories(),
 432           cfd->GetID(), cfd->GetName(), {}, kMaxSequenceNumber,
 433           snapshot_checker, kNoCompression, 0 /* sample_for_compression */,
 434           CompressionOptions(), false, nullptr /* internal_stats */,
 435           TableFileCreationReason::kRecovery, nullptr /* event_logger */,
 436           0 /* job_id */, Env::IO_HIGH, nullptr /* table_properties */,
 437           -1 /* level */, current_time, write_hint);
 438       ROCKS_LOG_INFO(db_options_.info_log,
 439                      "Log #%" PRIu64 ": %d ops saved to Table #%" PRIu64 " %s",
 440                      log, counter, meta.fd.GetNumber(),
 441                      status.ToString().c_str());
 442       if (status.ok()) {
 443         if (meta.fd.GetFileSize() > 0) {
 444           table_fds_.push_back(meta.fd);
 445         }
 446       } else {
 447         break;
 448       }
 449     }
 450     delete cf_mems;
 451     return status;
 452   }
 453
 454   void ExtractMetaData() {
 455     for (size_t i = 0; i < table_fds_.size(); i++) {
 456       TableInfo t;
 457       t.meta.fd = table_fds_[i];
 458       Status status = ScanTable(&t);
 459       if (!status.ok()) {
 460         std::string fname = TableFileName(
 461             db_options_.db_paths, t.meta.fd.GetNumber(), t.meta.fd.GetPathId());
 462         char file_num_buf[kFormatFileNumberBufSize];
 463         FormatFileNumber(t.meta.fd.GetNumber(), t.meta.fd.GetPathId(),
 464                          file_num_buf, sizeof(file_num_buf));
 465         ROCKS_LOG_WARN(db_options_.info_log, "Table #%s: ignoring %s",
 466                        file_num_buf, status.ToString().c_str());
 467         ArchiveFile(fname);
 468       } else {
 469         tables_.push_back(t);
 470       }
 471     }
 472   }
 473
 474   Status ScanTable(TableInfo* t) {
 475     std::string fname = TableFileName(
 476         db_options_.db_paths, t->meta.fd.GetNumber(), t->meta.fd.GetPathId());
 477     int counter = 0;
 478     uint64_t file_size;
 479     Status status = env_->GetFileSize(fname, &file_size);
 480     t->meta.fd = FileDescriptor(t->meta.fd.GetNumber(), t->meta.fd.GetPathId(),
 481                                 file_size);
 482     std::shared_ptr<const TableProperties> props;
 483     if (status.ok()) {
 484       status = table_cache_->GetTableProperties(env_options_, icmp_, t->meta.fd,
 485                                                 &props);
 486     }
 487     if (status.ok()) {
 488       t->column_family_id = static_cast<uint32_t>(props->column_family_id);
 489       if (t->column_family_id ==
 490           TablePropertiesCollectorFactory::Context::kUnknownColumnFamily) {
 491         ROCKS_LOG_WARN(
 492             db_options_.info_log,
 493             "Table #%" PRIu64
 494             ": column family unknown (probably due to legacy format); "
 495             "adding to default column family id 0.",
 496             t->meta.fd.GetNumber());
 497         t->column_family_id = 0;
 498       }
 499
 500       if (vset_.GetColumnFamilySet()->GetColumnFamily(t->column_family_id) ==
 501           nullptr) {
 502         status =
 503             AddColumnFamily(props->column_family_name, t->column_family_id);
 504       }
 505       t->meta.oldest_ancester_time = props->creation_time;
 506     }
 507     ColumnFamilyData* cfd = nullptr;
 508     if (status.ok()) {
 509       cfd = vset_.GetColumnFamilySet()->GetColumnFamily(t->column_family_id);
 510       if (cfd->GetName() != props->column_family_name) {
 511         ROCKS_LOG_ERROR(
 512             db_options_.info_log,
 513             "Table #%" PRIu64
 514             ": inconsistent column family name '%s'; expected '%s' for column "
 515             "family id %" PRIu32 ".",
 516             t->meta.fd.GetNumber(), props->column_family_name.c_str(),
 517             cfd->GetName().c_str(), t->column_family_id);
 518         status = Status::Corruption(dbname_, "inconsistent column family name");
 519       }
 520     }
 521     if (status.ok()) {
 522       ReadOptions ropts;
 523       ropts.total_order_seek = true;
 524       InternalIterator* iter = table_cache_->NewIterator(
 525           ropts, env_options_, cfd->internal_comparator(), t->meta,
 526           nullptr /* range_del_agg */,
 527           cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
 528           /*table_reader_ptr=*/nullptr, /*file_read_hist=*/nullptr,
 529           TableReaderCaller::kRepair, /*arena=*/nullptr, /*skip_filters=*/false,
 530           /*level=*/-1, /*smallest_compaction_key=*/nullptr,
 531           /*largest_compaction_key=*/nullptr);
 532       ParsedInternalKey parsed;
 533       for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
 534         Slice key = iter->key();
 535         if (!ParseInternalKey(key, &parsed)) {
 536           ROCKS_LOG_ERROR(db_options_.info_log,
 537                           "Table #%" PRIu64 ": unparsable key %s",
 538                           t->meta.fd.GetNumber(), EscapeString(key).c_str());
 539           continue;
 540         }
 541
 542         counter++;
 543
 544         t->meta.UpdateBoundaries(key, iter->value(), parsed.sequence,
 545                                  parsed.type);
 546       }
 547       if (!iter->status().ok()) {
 548         status = iter->status();
 549       }
 550       delete iter;
 551
 552       ROCKS_LOG_INFO(db_options_.info_log, "Table #%" PRIu64 ": %d entries %s",
 553                      t->meta.fd.GetNumber(), counter,
 554                      status.ToString().c_str());
 555     }
 556     return status;
 557   }
 558
 559   Status AddTables() {
 560     std::unordered_map<uint32_t, std::vector<const TableInfo*>> cf_id_to_tables;
 561     SequenceNumber max_sequence = 0;
 562     for (size_t i = 0; i < tables_.size(); i++) {
 563       cf_id_to_tables[tables_[i].column_family_id].push_back(&tables_[i]);
 564       if (max_sequence < tables_[i].meta.fd.largest_seqno) {
 565         max_sequence = tables_[i].meta.fd.largest_seqno;
 566       }
 567     }
 568     vset_.SetLastAllocatedSequence(max_sequence);
 569     vset_.SetLastPublishedSequence(max_sequence);
 570     vset_.SetLastSequence(max_sequence);
 571
 572     for (const auto& cf_id_and_tables : cf_id_to_tables) {
 573       auto* cfd =
 574           vset_.GetColumnFamilySet()->GetColumnFamily(cf_id_and_tables.first);
 575       VersionEdit edit;
 576       edit.SetComparatorName(cfd->user_comparator()->Name());
 577       edit.SetLogNumber(0);
 578       edit.SetNextFile(next_file_number_);
 579       edit.SetColumnFamily(cfd->GetID());
 580
 581       // TODO(opt): separate out into multiple levels
 582       for (const auto* table : cf_id_and_tables.second) {
 583         edit.AddFile(
 584             0, table->meta.fd.GetNumber(), table->meta.fd.GetPathId(),
 585             table->meta.fd.GetFileSize(), table->meta.smallest,
 586             table->meta.largest, table->meta.fd.smallest_seqno,
 587             table->meta.fd.largest_seqno, table->meta.marked_for_compaction,
 588             table->meta.oldest_blob_file_number,
 589             table->meta.oldest_ancester_time, table->meta.file_creation_time,
 590             table->meta.file_checksum, table->meta.file_checksum_func_name);
 591       }
 592       assert(next_file_number_ > 0);
 593       vset_.MarkFileNumberUsed(next_file_number_ - 1);
 594       mutex_.Lock();
 595       Status status = vset_.LogAndApply(
 596           cfd, *cfd->GetLatestMutableCFOptions(), &edit, &mutex_,
 597           nullptr /* db_directory */, false /* new_descriptor_log */);
 598       mutex_.Unlock();
 599       if (!status.ok()) {
 600         return status;
 601       }
 602     }
 603     return Status::OK();
 604   }
 605
 606   void ArchiveFile(const std::string& fname) {
 607     // Move into another directory.  E.g., for
 608     //    dir/foo
 609     // rename to
 610     //    dir/lost/foo
 611     const char* slash = strrchr(fname.c_str(), '/');
 612     std::string new_dir;
 613     if (slash != nullptr) {
 614       new_dir.assign(fname.data(), slash - fname.data());
 615     }
 616     new_dir.append("/lost");
 617     env_->CreateDir(new_dir);  // Ignore error
 618     std::string new_file = new_dir;
 619     new_file.append("/");
 620     new_file.append((slash == nullptr) ? fname.c_str() : slash + 1);
 621     Status s = env_->RenameFile(fname, new_file);
 622     ROCKS_LOG_INFO(db_options_.info_log, "Archiving %s: %s\n", fname.c_str(),
 623                    s.ToString().c_str());
 624   }
 625 };
 626
 627 Status GetDefaultCFOptions(
 628     const std::vector<ColumnFamilyDescriptor>& column_families,
 629     ColumnFamilyOptions* res) {
 630   assert(res != nullptr);
 631   auto iter = std::find_if(column_families.begin(), column_families.end(),
 632                            [](const ColumnFamilyDescriptor& cfd) {
 633                              return cfd.name == kDefaultColumnFamilyName;
 634                            });
 635   if (iter == column_families.end()) {
 636     return Status::InvalidArgument(
 637         "column_families", "Must contain entry for default column family");
 638   }
 639   *res = iter->options;
 640   return Status::OK();
 641 }
 642 }  // anonymous namespace
 643
 644 Status RepairDB(const std::string& dbname, const DBOptions& db_options,
 645                 const std::vector<ColumnFamilyDescriptor>& column_families
 646                 ) {
 647   ColumnFamilyOptions default_cf_opts;
 648   Status status = GetDefaultCFOptions(column_families, &default_cf_opts);
 649   if (status.ok()) {
 650     Repairer repairer(dbname, db_options, column_families,
 651                       default_cf_opts,
 652                       ColumnFamilyOptions() /* unknown_cf_opts */,
 653                       false /* create_unknown_cfs */);
 654     status = repairer.Run();
 655   }
 656   return status;
 657 }
 658
 659 Status RepairDB(const std::string& dbname, const DBOptions& db_options,
 660                 const std::vector<ColumnFamilyDescriptor>& column_families,
 661                 const ColumnFamilyOptions& unknown_cf_opts) {
 662   ColumnFamilyOptions default_cf_opts;
 663   Status status = GetDefaultCFOptions(column_families, &default_cf_opts);
 664   if (status.ok()) {
 665     Repairer repairer(dbname, db_options,
 666                       column_families, default_cf_opts,
 667                       unknown_cf_opts, true /* create_unknown_cfs */);
 668     status = repairer.Run();
 669   }
 670   return status;
 671 }
 672
 673 Status RepairDB(const std::string& dbname, const Options& options) {
 674   Options opts(options);
 675   if (opts.file_system == nullptr) {
 676     opts.file_system.reset(new LegacyFileSystemWrapper(opts.env));
 677     ;
 678   }
 679
 680   DBOptions db_options(opts);
 681   ColumnFamilyOptions cf_options(opts);
 682   Repairer repairer(dbname, db_options,
 683                     {}, cf_options /* default_cf_opts */,
 684                     cf_options /* unknown_cf_opts */,
 685                     true /* create_unknown_cfs */);
 686   return repairer.Run();
 687 }
 688
 689 }  // namespace ROCKSDB_NAMESPACE
 690
 691 #endif  // ROCKSDB_LITE