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

#pragma once

#include <atomic>

#include <cstdint>

#include <deque>

#include <functional>

#include <limits>

#include <list>

#include <map>

#include <set>

#include <string>

#include <unordered_map>

#include <utility>

#include <vector>

#include "db/column_family.h"

#include "db/compaction/compaction_iterator.h"

#include "db/compaction/compaction_job.h"

#include "db/error_handler.h"

#include "db/event_helpers.h"

#include "db/external_sst_file_ingestion_job.h"

#include "db/flush_job.h"

#include "db/flush_scheduler.h"

#include "db/import_column_family_job.h"

#include "db/internal_stats.h"

#include "db/log_writer.h"

#include "db/logs_with_prep_tracker.h"

#include "db/memtable_list.h"

#include "db/periodic_task_scheduler.h"

#include "db/post_memtable_callback.h"

#include "db/pre_release_callback.h"

#include "db/range_del_aggregator.h"

#include "db/read_callback.h"

#include "db/seqno_to_time_mapping.h"

#include "db/snapshot_checker.h"

#include "db/snapshot_impl.h"

#include "db/trim_history_scheduler.h"

#include "db/version_edit.h"

#include "db/wal_manager.h"

#include "db/write_controller.h"

#include "db/write_thread.h"

#include "logging/event_logger.h"

#include "monitoring/instrumented_mutex.h"

#include "options/db_options.h"

#include "port/port.h"

#include "rocksdb/attribute_groups.h"

#include "rocksdb/db.h"

#include "rocksdb/env.h"

#include "rocksdb/memtablerep.h"

#include "rocksdb/status.h"

#include "rocksdb/trace_reader_writer.h"

#include "rocksdb/transaction_log.h"

#include "rocksdb/user_write_callback.h"

#include "rocksdb/utilities/replayer.h"

#include "rocksdb/write_buffer_manager.h"

#include "table/merging_iterator.h"

#include "util/autovector.h"

#include "util/hash.h"

#include "util/repeatable_thread.h"

#include "util/stop_watch.h"

#include "util/thread_local.h"

namespace ROCKSDB_NAMESPACE {

class Arena;

class ArenaWrappedDBIter;

class InMemoryStatsHistoryIterator;

class MemTable;

class PersistentStatsHistoryIterator;

class TableCache;

class TaskLimiterToken;

class Version;

class VersionEdit;

class VersionSet;

class WriteCallback;

struct JobContext;

struct ExternalSstFileInfo;

struct MemTableInfo;

// Class to maintain directories for all database paths other than main one.

class Directories {

 public:

  IOStatus SetDirectories(FileSystem* fs, const std::string& dbname,

                          const std::string& wal_dir,

                          const std::vector<DbPath>& data_paths);

  FSDirectory* GetDataDir(size_t path_id) const {

    assert(path_id < data_dirs_.size());

    FSDirectory* ret_dir = data_dirs_[path_id].get();

    if (ret_dir == nullptr) {

      // Should use db_dir_

      return db_dir_.get();

    }

    return ret_dir;

  }

  FSDirectory* GetWalDir() {

    if (wal_dir_) {

      return wal_dir_.get();

    }

    return db_dir_.get();

  }

  FSDirectory* GetDbDir() { return db_dir_.get(); }

  IOStatus Close(const IOOptions& options, IODebugContext* dbg) {

    // close all directories for all database paths

    IOStatus s = IOStatus::OK();

    // The default implementation for Close() in Directory/FSDirectory class

    // "NotSupported" status, the upper level interface should be able to

    // handle this error so that Close() does not fail after upgrading when

    // run on FileSystems that have not implemented `Directory::Close()` or

    // `FSDirectory::Close()` yet

    if (db_dir_) {

      IOStatus temp_s = db_dir_->Close(options, dbg);

      if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {

        s = std::move(temp_s);

      }

    }

    // Attempt to close everything even if one fails

    s.PermitUncheckedError();

    if (wal_dir_) {

      IOStatus temp_s = wal_dir_->Close(options, dbg);

      if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {

        s = std::move(temp_s);

      }

    }

    s.PermitUncheckedError();

    for (auto& data_dir_ptr : data_dirs_) {

      if (data_dir_ptr) {

        IOStatus temp_s = data_dir_ptr->Close(options, dbg);

        if (!temp_s.ok() && !temp_s.IsNotSupported() && s.ok()) {

          s = std::move(temp_s);

        }

      }

    }

    // Ready for caller

    s.MustCheck();

    return s;

  }

 private:

  std::unique_ptr<FSDirectory> db_dir_;

  std::vector<std::unique_ptr<FSDirectory>> data_dirs_;

  std::unique_ptr<FSDirectory> wal_dir_;

};

// While DB is the public interface of RocksDB, and DBImpl is the actual

// class implementing it. It's the entrance of the core RocksdB engine.

// All other DB implementations, e.g. TransactionDB, BlobDB, etc, wrap a

// DBImpl internally.

// Other than functions implementing the DB interface, some public

// functions are there for other internal components to call. For

// example, TransactionDB directly calls DBImpl::WriteImpl() and

// BlobDB directly calls DBImpl::GetImpl(). Some other functions

// are for sub-components to call. For example, ColumnFamilyHandleImpl

// calls DBImpl::FindObsoleteFiles().

//

// Since it's a very large class, the definition of the functions is

// divided in several db_impl_*.cc files, besides db_impl.cc.

class DBImpl : public DB {

 public:

  DBImpl(const DBOptions& options, const std::string& dbname,

         const bool seq_per_batch = false, const bool batch_per_txn = true,

         bool read_only = false);

  // No copying allowed

  DBImpl(const DBImpl&) = delete;

  void operator=(const DBImpl&) = delete;

  virtual ~DBImpl();

  // ---- Implementations of the DB interface ----

  using DB::Resume;

  Status Resume() override;

  using DB::Put;

  Status Put(const WriteOptions& options, ColumnFamilyHandle* column_family,

             const Slice& key, const Slice& value) override;

  Status Put(const WriteOptions& options, ColumnFamilyHandle* column_family,

             const Slice& key, const Slice& ts, const Slice& value) override;

  using DB::PutEntity;

  Status PutEntity(const WriteOptions& options,

                   ColumnFamilyHandle* column_family, const Slice& key,

                   const WideColumns& columns) override;

  Status PutEntity(const WriteOptions& options, const Slice& key,

                   const AttributeGroups& attribute_groups) override;

  using DB::Merge;

  Status Merge(const WriteOptions& options, ColumnFamilyHandle* column_family,

               const Slice& key, const Slice& value) override;

  Status Merge(const WriteOptions& options, ColumnFamilyHandle* column_family,

               const Slice& key, const Slice& ts, const Slice& value) override;

  using DB::Delete;

  Status Delete(const WriteOptions& options, ColumnFamilyHandle* column_family,

                const Slice& key) override;

  Status Delete(const WriteOptions& options, ColumnFamilyHandle* column_family,

                const Slice& key, const Slice& ts) override;

  using DB::SingleDelete;

  Status SingleDelete(const WriteOptions& options,

                      ColumnFamilyHandle* column_family,

                      const Slice& key) override;

  Status SingleDelete(const WriteOptions& options,

                      ColumnFamilyHandle* column_family, const Slice& key,

                      const Slice& ts) override;

  using DB::DeleteRange;

  Status DeleteRange(const WriteOptions& options,

                     ColumnFamilyHandle* column_family, const Slice& begin_key,

                     const Slice& end_key) override;

  Status DeleteRange(const WriteOptions& options,

                     ColumnFamilyHandle* column_family, const Slice& begin_key,

                     const Slice& end_key, const Slice& ts) override;

  using DB::Write;

  Status Write(const WriteOptions& options, WriteBatch* updates) override;

  using DB::WriteWithCallback;

  Status WriteWithCallback(const WriteOptions& options, WriteBatch* updates,

                           UserWriteCallback* user_write_cb) override;

  using DB::Get;

  Status Get(const ReadOptions& _read_options,

             ColumnFamilyHandle* column_family, const Slice& key,

             PinnableSlice* value, std::string* timestamp) override;

  using DB::GetEntity;

  Status GetEntity(const ReadOptions& options,

                   ColumnFamilyHandle* column_family, const Slice& key,

                   PinnableWideColumns* columns) override;

  Status GetEntity(const ReadOptions& options, const Slice& key,

                   PinnableAttributeGroups* result) override;

  using DB::GetMergeOperands;

  Status GetMergeOperands(const ReadOptions& options,

                          ColumnFamilyHandle* column_family, const Slice& key,

                          PinnableSlice* merge_operands,

                          GetMergeOperandsOptions* get_merge_operands_options,

                          int* number_of_operands) override {

    GetImplOptions get_impl_options;

    get_impl_options.column_family = column_family;

    get_impl_options.merge_operands = merge_operands;

    get_impl_options.get_merge_operands_options = get_merge_operands_options;

    get_impl_options.number_of_operands = number_of_operands;

    get_impl_options.get_value = false;

    return GetImpl(options, key, get_impl_options);

  }

  using DB::MultiGet;

  // This MultiGet is a batched version, which may be faster than calling Get

  // multiple times, especially if the keys have some spatial locality that

  // enables them to be queried in the same SST files/set of files. The larger

  // the batch size, the more scope for batching and performance improvement

  // The values and statuses parameters are arrays with number of elements

  // equal to keys.size(). This allows the storage for those to be alloacted

  // by the caller on the stack for small batches

  void MultiGet(const ReadOptions& _read_options, const size_t num_keys,

                ColumnFamilyHandle** column_families, const Slice* keys,

                PinnableSlice* values, std::string* timestamps,

                Status* statuses, const bool sorted_input = false) override;

  void MultiGetWithCallback(

      const ReadOptions& _read_options, ColumnFamilyHandle* column_family,

      ReadCallback* callback,

      autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE>* sorted_keys);

  using DB::MultiGetEntity;

  void MultiGetEntity(const ReadOptions& options,

                      ColumnFamilyHandle* column_family, size_t num_keys,

                      const Slice* keys, PinnableWideColumns* results,

                      Status* statuses, bool sorted_input) override;

  void MultiGetEntity(const ReadOptions& options, size_t num_keys,

                      ColumnFamilyHandle** column_families, const Slice* keys,

                      PinnableWideColumns* results, Status* statuses,

                      bool sorted_input) override;

  void MultiGetEntity(const ReadOptions& options, size_t num_keys,

                      const Slice* keys,

                      PinnableAttributeGroups* results) override;

  void MultiGetEntityWithCallback(

      const ReadOptions& read_options, ColumnFamilyHandle* column_family,

      ReadCallback* callback,

      autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE>* sorted_keys);

  Status CreateColumnFamily(const ColumnFamilyOptions& cf_options,

                            const std::string& column_family,

                            ColumnFamilyHandle** handle) override {

    // TODO: plumb Env::IOActivity, Env::IOPriority

    return CreateColumnFamily(ReadOptions(), WriteOptions(), cf_options,

                              column_family, handle);

  }

  virtual Status CreateColumnFamily(const ReadOptions& read_options,

                                    const WriteOptions& write_options,

                                    const ColumnFamilyOptions& cf_options,

                                    const std::string& column_family,

                                    ColumnFamilyHandle** handle);

  Status CreateColumnFamilies(

      const ColumnFamilyOptions& cf_options,

      const std::vector<std::string>& column_family_names,

      std::vector<ColumnFamilyHandle*>* handles) override {

    // TODO: plumb Env::IOActivity, Env::IOPriority

    return CreateColumnFamilies(ReadOptions(), WriteOptions(), cf_options,

                                column_family_names, handles);

  }

  virtual Status CreateColumnFamilies(

      const ReadOptions& read_options, const WriteOptions& write_options,

      const ColumnFamilyOptions& cf_options,

      const std::vector<std::string>& column_family_names,

      std::vector<ColumnFamilyHandle*>* handles);

  Status CreateColumnFamilies(

      const std::vector<ColumnFamilyDescriptor>& column_families,

      std::vector<ColumnFamilyHandle*>* handles) override {

    // TODO: plumb Env::IOActivity, Env::IOPriority

    return CreateColumnFamilies(ReadOptions(), WriteOptions(), column_families,

                                handles);

  }

  virtual Status CreateColumnFamilies(

      const ReadOptions& read_options, const WriteOptions& write_options,

      const std::vector<ColumnFamilyDescriptor>& column_families,

      std::vector<ColumnFamilyHandle*>* handles);

  Status DropColumnFamily(ColumnFamilyHandle* column_family) override;

  Status DropColumnFamilies(

      const std::vector<ColumnFamilyHandle*>& column_families) override;

  // Returns false if key doesn't exist in the database and true if it may.

  // If value_found is not passed in as null, then return the value if found in

  // memory. On return, if value was found, then value_found will be set to true

  // , otherwise false.

  using DB::KeyMayExist;

  bool KeyMayExist(const ReadOptions& options,

                   ColumnFamilyHandle* column_family, const Slice& key,

                   std::string* value, std::string* timestamp,

                   bool* value_found = nullptr) override;

  using DB::NewIterator;

  Iterator* NewIterator(const ReadOptions& _read_options,

                        ColumnFamilyHandle* column_family) override;

  Status NewIterators(const ReadOptions& _read_options,

                      const std::vector<ColumnFamilyHandle*>& column_families,

                      std::vector<Iterator*>* iterators) override;

  const Snapshot* GetSnapshot() override;

  void ReleaseSnapshot(const Snapshot* snapshot) override;

  // EXPERIMENTAL

  std::unique_ptr<Iterator> NewCoalescingIterator(

      const ReadOptions& options,

      const std::vector<ColumnFamilyHandle*>& column_families) override;

  // EXPERIMENTAL

  std::unique_ptr<AttributeGroupIterator> NewAttributeGroupIterator(

      const ReadOptions& options,

      const std::vector<ColumnFamilyHandle*>& column_families) override;

  // Create a timestamped snapshot. This snapshot can be shared by multiple

  // readers. If any of them uses it for write conflict checking, then

  // is_write_conflict_boundary is true. For simplicity, set it to true by

  // default.

  std::pair<Status, std::shared_ptr<const Snapshot>> CreateTimestampedSnapshot(

      SequenceNumber snapshot_seq, uint64_t ts);

  std::shared_ptr<const SnapshotImpl> GetTimestampedSnapshot(uint64_t ts) const;

  void ReleaseTimestampedSnapshotsOlderThan(

      uint64_t ts, size_t* remaining_total_ss = nullptr);

  Status GetTimestampedSnapshots(uint64_t ts_lb, uint64_t ts_ub,

                                 std::vector<std::shared_ptr<const Snapshot>>&

                                     timestamped_snapshots) const;

  using DB::GetProperty;

  bool GetProperty(ColumnFamilyHandle* column_family, const Slice& property,

                   std::string* value) override;

  using DB::GetMapProperty;

  bool GetMapProperty(ColumnFamilyHandle* column_family, const Slice& property,

                      std::map<std::string, std::string>* value) override;

  using DB::GetIntProperty;

  bool GetIntProperty(ColumnFamilyHandle* column_family, const Slice& property,

                      uint64_t* value) override;

  using DB::GetAggregatedIntProperty;

  bool GetAggregatedIntProperty(const Slice& property,

                                uint64_t* aggregated_value) override;

  using DB::GetApproximateSizes;

  Status GetApproximateSizes(const SizeApproximationOptions& options,

                             ColumnFamilyHandle* column_family,

                             const Range* range, int n,

                             uint64_t* sizes) override;

  using DB::GetApproximateMemTableStats;

  void GetApproximateMemTableStats(ColumnFamilyHandle* column_family,

                                   const Range& range, uint64_t* const count,

                                   uint64_t* const size) override;

  using DB::CompactRange;

  Status CompactRange(const CompactRangeOptions& options,

                      ColumnFamilyHandle* column_family, const Slice* begin,

                      const Slice* end) override;

  using DB::CompactFiles;

  Status CompactFiles(

      const CompactionOptions& compact_options,

      ColumnFamilyHandle* column_family,

      const std::vector<std::string>& input_file_names, const int output_level,

      const int output_path_id = -1,

      std::vector<std::string>* const output_file_names = nullptr,

      CompactionJobInfo* compaction_job_info = nullptr) override;

  Status PauseBackgroundWork() override;

  Status ContinueBackgroundWork() override;

  Status EnableAutoCompaction(

      const std::vector<ColumnFamilyHandle*>& column_family_handles) override;

  void EnableManualCompaction() override;

  void DisableManualCompaction() override;

  using DB::SetOptions;

  Status SetOptions(

      ColumnFamilyHandle* column_family,

      const std::unordered_map<std::string, std::string>& options_map) override;

  Status SetDBOptions(

      const std::unordered_map<std::string, std::string>& options_map) override;

  using DB::NumberLevels;

  int NumberLevels(ColumnFamilyHandle* column_family) override;

  using DB::MaxMemCompactionLevel;

  int MaxMemCompactionLevel(ColumnFamilyHandle* column_family) override;

  using DB::Level0StopWriteTrigger;

  int Level0StopWriteTrigger(ColumnFamilyHandle* column_family) override;

  const std::string& GetName() const override;

  Env* GetEnv() const override;

  FileSystem* GetFileSystem() const override;

  using DB::GetOptions;

  Options GetOptions(ColumnFamilyHandle* column_family) const override;

  using DB::GetDBOptions;

  DBOptions GetDBOptions() const override;

  using DB::Flush;

  Status Flush(const FlushOptions& options,

               ColumnFamilyHandle* column_family) override;

  Status Flush(

      const FlushOptions& options,

      const std::vector<ColumnFamilyHandle*>& column_families) override;

  Status FlushWAL(bool sync) override {

    // TODO: plumb Env::IOActivity, Env::IOPriority

    return FlushWAL(WriteOptions(), sync);

  }

  virtual Status FlushWAL(const WriteOptions& write_options, bool sync);

  bool WALBufferIsEmpty();

  Status SyncWAL() override;

  Status LockWAL() override;

  Status UnlockWAL() override;

  SequenceNumber GetLatestSequenceNumber() const override;

  // IncreaseFullHistoryTsLow(ColumnFamilyHandle*, std::string) will acquire

  // and release db_mutex

  Status IncreaseFullHistoryTsLow(ColumnFamilyHandle* column_family,

                                  std::string ts_low) override;

  // GetFullHistoryTsLow(ColumnFamilyHandle*, std::string*) will acquire and

  // release db_mutex

  Status GetFullHistoryTsLow(ColumnFamilyHandle* column_family,

                             std::string* ts_low) override;

  Status GetDbIdentity(std::string& identity) const override;

  virtual Status GetDbIdentityFromIdentityFile(std::string* identity) const;

  Status GetDbSessionId(std::string& session_id) const override;

  ColumnFamilyHandle* DefaultColumnFamily() const override;

  ColumnFamilyHandle* PersistentStatsColumnFamily() const;

  Status Close() override;

  Status DisableFileDeletions() override;

  Status EnableFileDeletions() override;

  virtual bool IsFileDeletionsEnabled() const;

  Status GetStatsHistory(

      uint64_t start_time, uint64_t end_time,

      std::unique_ptr<StatsHistoryIterator>* stats_iterator) override;

  using DB::ResetStats;

  Status ResetStats() override;

  // All the returned filenames start with "/"

  Status GetLiveFiles(std::vector<std::string>&, uint64_t* manifest_file_size,

                      bool flush_memtable = true) override;

  Status GetSortedWalFiles(VectorWalPtr& files) override;

  Status GetSortedWalFilesImpl(VectorWalPtr& files, bool need_seqnos);

  // Get the known flushed sizes of WALs that might still be written to

  // or have pending sync.

  // NOTE: unlike alive_log_files_, this function includes WALs that might

  // be obsolete (but not obsolete to a pending Checkpoint) and not yet fully

  // synced.

  Status GetOpenWalSizes(std::map<uint64_t, uint64_t>& number_to_size);

  Status GetCurrentWalFile(std::unique_ptr<WalFile>* current_log_file) override;

  Status GetCreationTimeOfOldestFile(uint64_t* creation_time) override;

  Status GetUpdatesSince(

      SequenceNumber seq_number, std::unique_ptr<TransactionLogIterator>* iter,

      const TransactionLogIterator::ReadOptions& read_options =

          TransactionLogIterator::ReadOptions()) override;

  Status DeleteFile(std::string name) override;

  Status DeleteFilesInRanges(ColumnFamilyHandle* column_family,

                             const RangePtr* ranges, size_t n,

                             bool include_end = true);

  void GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata) override;

  Status GetLiveFilesChecksumInfo(FileChecksumList* checksum_list) override;

  Status GetLiveFilesStorageInfo(

      const LiveFilesStorageInfoOptions& opts,

      std::vector<LiveFileStorageInfo>* files) override;

  // Obtains the meta data of the specified column family of the DB.

  // TODO(yhchiang): output parameter is placed in the end in this codebase.

  void GetColumnFamilyMetaData(ColumnFamilyHandle* column_family,

                               ColumnFamilyMetaData* metadata) override;

  void GetAllColumnFamilyMetaData(

      std::vector<ColumnFamilyMetaData>* metadata) override;

  Status SuggestCompactRange(ColumnFamilyHandle* column_family,

                             const Slice* begin, const Slice* end) override;

  Status PromoteL0(ColumnFamilyHandle* column_family,

                   int target_level) override;

  using DB::IngestExternalFile;

  Status IngestExternalFile(

      ColumnFamilyHandle* column_family,

      const std::vector<std::string>& external_files,

      const IngestExternalFileOptions& ingestion_options) override;

  using DB::IngestExternalFiles;

  Status IngestExternalFiles(

      const std::vector<IngestExternalFileArg>& args) override;

  using DB::CreateColumnFamilyWithImport;

  Status CreateColumnFamilyWithImport(

      const ColumnFamilyOptions& options, const std::string& column_family_name,

      const ImportColumnFamilyOptions& import_options,

      const std::vector<const ExportImportFilesMetaData*>& metadatas,

      ColumnFamilyHandle** handle) override;

  using DB::ClipColumnFamily;

  Status ClipColumnFamily(ColumnFamilyHandle* column_family,

                          const Slice& begin_key,

                          const Slice& end_key) override;

  using DB::VerifyFileChecksums;

  Status VerifyFileChecksums(const ReadOptions& read_options) override;

  using DB::VerifyChecksum;

  Status VerifyChecksum(const ReadOptions& /*read_options*/) override;

  // Verify the checksums of files in db. Currently only tables are checked.

  //

  // read_options: controls file I/O behavior, e.g. read ahead size while

  //               reading all the live table files.

  //

  // use_file_checksum: if false, verify the block checksums of all live table

  //                    in db. Otherwise, obtain the file checksums and compare

  //                    with the MANIFEST. Currently, file checksums are

  //                    recomputed by reading all table files.

  //

  // Returns: OK if there is no file whose file or block checksum mismatches.

  Status VerifyChecksumInternal(const ReadOptions& read_options,

                                bool use_file_checksum);

  Status VerifyFullFileChecksum(const std::string& file_checksum_expected,

                                const std::string& func_name_expected,

                                const std::string& fpath,

                                const ReadOptions& read_options);

  using DB::StartTrace;

  Status StartTrace(const TraceOptions& options,

                    std::unique_ptr<TraceWriter>&& trace_writer) override;

  using DB::EndTrace;

  Status EndTrace() override;

  using DB::NewDefaultReplayer;

  Status NewDefaultReplayer(const std::vector<ColumnFamilyHandle*>& handles,

                            std::unique_ptr<TraceReader>&& reader,

                            std::unique_ptr<Replayer>* replayer) override;

  using DB::StartBlockCacheTrace;

  Status StartBlockCacheTrace(

      const TraceOptions& trace_options,

      std::unique_ptr<TraceWriter>&& trace_writer) override;

  Status StartBlockCacheTrace(

      const BlockCacheTraceOptions& options,

      std::unique_ptr<BlockCacheTraceWriter>&& trace_writer) override;

  using DB::EndBlockCacheTrace;

  Status EndBlockCacheTrace() override;

  using DB::StartIOTrace;

  Status StartIOTrace(const TraceOptions& options,

                      std::unique_ptr<TraceWriter>&& trace_writer) override;

  using DB::EndIOTrace;

  Status EndIOTrace() override;

  using DB::GetPropertiesOfAllTables;

  Status GetPropertiesOfAllTables(ColumnFamilyHandle* column_family,

                                  TablePropertiesCollection* props) override;

  Status GetPropertiesOfTablesInRange(

      ColumnFamilyHandle* column_family, const Range* range, std::size_t n,

      TablePropertiesCollection* props) override;

  // ---- End of implementations of the DB interface ----

  SystemClock* GetSystemClock() const;

  struct GetImplOptions {

    ColumnFamilyHandle* column_family = nullptr;

    PinnableSlice* value = nullptr;

    PinnableWideColumns* columns = nullptr;

    std::string* timestamp = nullptr;

    bool* value_found = nullptr;

    ReadCallback* callback = nullptr;

    bool* is_blob_index = nullptr;

    // If true return value associated with key via value pointer else return

    // all merge operands for key via merge_operands pointer

    bool get_value = true;

    // Pointer to an array of size

    // get_merge_operands_options.expected_max_number_of_operands allocated by

    // user

    PinnableSlice* merge_operands = nullptr;

    GetMergeOperandsOptions* get_merge_operands_options = nullptr;

    int* number_of_operands = nullptr;

  };

  Status GetImpl(const ReadOptions& read_options,

                 ColumnFamilyHandle* column_family, const Slice& key,

                 PinnableSlice* value);

  Status GetImpl(const ReadOptions& read_options,

                 ColumnFamilyHandle* column_family, const Slice& key,

                 PinnableSlice* value, std::string* timestamp);

  // Function that Get and KeyMayExist call with no_io true or false

  // Note: 'value_found' from KeyMayExist propagates here

  // This function is also called by GetMergeOperands

  // If get_impl_options.get_value = true get value associated with

  // get_impl_options.key via get_impl_options.value

  // If get_impl_options.get_value = false get merge operands associated with

  // get_impl_options.key via get_impl_options.merge_operands

  virtual Status GetImpl(const ReadOptions& options, const Slice& key,

                         GetImplOptions& get_impl_options);

  // If `snapshot` == kMaxSequenceNumber, set a recent one inside the file.

  ArenaWrappedDBIter* NewIteratorImpl(const ReadOptions& options,

                                      ColumnFamilyHandleImpl* cfh,

                                      SuperVersion* sv, SequenceNumber snapshot,

                                      ReadCallback* read_callback,

                                      bool expose_blob_index = false,

                                      bool allow_refresh = true);

  virtual SequenceNumber GetLastPublishedSequence() const {

    if (last_seq_same_as_publish_seq_) {

      return versions_->LastSequence();

    } else {

      return versions_->LastPublishedSequence();

    }

  }

  // REQUIRES: joined the main write queue if two_write_queues is disabled, and

  // the second write queue otherwise.

  virtual void SetLastPublishedSequence(SequenceNumber seq);

  // Returns LastSequence in last_seq_same_as_publish_seq_

  // mode and LastAllocatedSequence otherwise. This is useful when visiblility

  // depends also on data written to the WAL but not to the memtable.

  SequenceNumber TEST_GetLastVisibleSequence() const;

  // Similar to Write() but will call the callback once on the single write

  // thread to determine whether it is safe to perform the write.

  virtual Status WriteWithCallback(const WriteOptions& write_options,

                                   WriteBatch* my_batch,

                                   WriteCallback* callback,

                                   UserWriteCallback* user_write_cb = nullptr);

  // Returns the sequence number that is guaranteed to be smaller than or equal

  // to the sequence number of any key that could be inserted into the current

  // memtables. It can then be assumed that any write with a larger(or equal)

  // sequence number will be present in this memtable or a later memtable.

  //

  // If the earliest sequence number could not be determined,

  // kMaxSequenceNumber will be returned.

  //

  // If include_history=true, will also search Memtables in MemTableList

  // History.

  SequenceNumber GetEarliestMemTableSequenceNumber(SuperVersion* sv,

                                                   bool include_history);

  // For a given key, check to see if there are any records for this key

  // in the memtables, including memtable history.  If cache_only is false,

  // SST files will also be checked.

  //

  // `key` should NOT have user-defined timestamp appended to user key even if

  // timestamp is enabled.

  //

  // If a key is found, *found_record_for_key will be set to true and

  // *seq will be set to the stored sequence number for the latest

  // operation on this key or kMaxSequenceNumber if unknown. If user-defined

  // timestamp is enabled for this column family and timestamp is not nullptr,

  // then *timestamp will be set to the stored timestamp for the latest

  // operation on this key.

  // If no key is found, *found_record_for_key will be set to false.

  //

  // Note: If cache_only=false, it is possible for *seq to be set to 0 if

  // the sequence number has been cleared from the record.  If the caller is

  // holding an active db snapshot, we know the missing sequence must be less

  // than the snapshot's sequence number (sequence numbers are only cleared

  // when there are no earlier active snapshots).

  //

  // If NotFound is returned and found_record_for_key is set to false, then no

  // record for this key was found.  If the caller is holding an active db

  // snapshot, we know that no key could have existing after this snapshot

  // (since we do not compact keys that have an earlier snapshot).

  //

  // Only records newer than or at `lower_bound_seq` are guaranteed to be

  // returned. Memtables and files may not be checked if it only contains data

  // older than `lower_bound_seq`.

  //

  // Returns OK or NotFound on success,

  // other status on unexpected error.

  // TODO(andrewkr): this API need to be aware of range deletion operations

  Status GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,

                                 bool cache_only,

                                 SequenceNumber lower_bound_seq,

                                 SequenceNumber* seq, std::string* timestamp,

                                 bool* found_record_for_key,

                                 bool* is_blob_index);

  Status TraceIteratorSeek(const uint32_t& cf_id, const Slice& key,

                           const Slice& lower_bound, const Slice upper_bound);

  Status TraceIteratorSeekForPrev(const uint32_t& cf_id, const Slice& key,

                                  const Slice& lower_bound,

                                  const Slice upper_bound);

  // Similar to GetSnapshot(), but also lets the db know that this snapshot

  // will be used for transaction write-conflict checking.  The DB can then

  // make sure not to compact any keys that would prevent a write-conflict from

  // being detected.

  const Snapshot* GetSnapshotForWriteConflictBoundary();

  // checks if all live files exist on file system and that their file sizes

  // match to our in-memory records

  virtual Status CheckConsistency();

  // max_file_num_to_ignore allows bottom level compaction to filter out newly

  // compacted SST files. Setting max_file_num_to_ignore to kMaxUint64 will

  // disable the filtering

  // If `final_output_level` is not nullptr, it is set to manual compaction's

  // output level if returned status is OK, and it may or may not be set to

  // manual compaction's output level if returned status is not OK.

  Status RunManualCompaction(ColumnFamilyData* cfd, int input_level,

                             int output_level,

                             const CompactRangeOptions& compact_range_options,

                             const Slice* begin, const Slice* end,

                             bool exclusive, bool disallow_trivial_move,

                             uint64_t max_file_num_to_ignore,

                             const std::string& trim_ts,

                             int* final_output_level = nullptr);

  // Return an internal iterator over the current state of the database.

  // The keys of this iterator are internal keys (see format.h).

  // The returned iterator should be deleted when no longer needed.

  // If allow_unprepared_value is true, the returned iterator may defer reading

  // the value and so will require PrepareValue() to be called before value();

  // allow_unprepared_value = false is convenient when this optimization is not

  // useful, e.g. when reading the whole column family.

  //

  // read_options.ignore_range_deletions determines whether range tombstones are

  // processed in the returned interator internally, i.e., whether range

  // tombstone covered keys are in this iterator's output.

  // @param read_options Must outlive the returned iterator.

  InternalIterator* NewInternalIterator(

      const ReadOptions& read_options, Arena* arena, SequenceNumber sequence,

      ColumnFamilyHandle* column_family = nullptr,

      bool allow_unprepared_value = false);

  // Note: to support DB iterator refresh, memtable range tombstones in the

  // underlying merging iterator needs to be refreshed. If db_iter is not

  // nullptr, db_iter->SetMemtableRangetombstoneIter() is called with the

  // memtable range tombstone iterator used by the underlying merging iterator.

  // This range tombstone iterator can be refreshed later by db_iter.

  // @param read_options Must outlive the returned iterator.

  InternalIterator* NewInternalIterator(const ReadOptions& read_options,

                                        ColumnFamilyData* cfd,

                                        SuperVersion* super_version,

                                        Arena* arena, SequenceNumber sequence,

                                        bool allow_unprepared_value,

                                        ArenaWrappedDBIter* db_iter = nullptr);

  LogsWithPrepTracker* logs_with_prep_tracker() {

    return &logs_with_prep_tracker_;

  }

  struct BGJobLimits {

    int max_flushes;

    int max_compactions;

  };

  // Returns maximum background flushes and compactions allowed to be scheduled

  BGJobLimits GetBGJobLimits() const;

  // Need a static version that can be called during SanitizeOptions().

  static BGJobLimits GetBGJobLimits(int max_background_flushes,

                                    int max_background_compactions,

                                    int max_background_jobs,

                                    bool parallelize_compactions);

  // move logs pending closing from job_context to the DB queue and

  // schedule a purge

  void ScheduleBgLogWriterClose(JobContext* job_context);

  uint64_t MinLogNumberToKeep();

  uint64_t MinLogNumberToRecycle();

  // Returns the lower bound file number for SSTs that won't be deleted, even if

  // they're obsolete. This lower bound is used internally to prevent newly

  // created flush/compaction output files from being deleted before they're

  // installed. This technique avoids the need for tracking the exact numbers of

  // files pending creation, although it prevents more files than necessary from

  // being deleted.

  uint64_t MinObsoleteSstNumberToKeep();

  uint64_t GetObsoleteSstFilesSize();

  // Returns the list of live files in 'live' and the list

  // of all files in the filesystem in 'candidate_files'.

  // If force == false and the last call was less than

  // db_options_.delete_obsolete_files_period_micros microseconds ago,

  // it will not fill up the job_context

  void FindObsoleteFiles(JobContext* job_context, bool force,

                         bool no_full_scan = false);

  // Diffs the files listed in filenames and those that do not

  // belong to live files are possibly removed. Also, removes all the

  // files in sst_delete_files and log_delete_files.

  // It is not necessary to hold the mutex when invoking this method.

  // If FindObsoleteFiles() was run, we need to also run

  // PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true

  void PurgeObsoleteFiles(JobContext& background_contet,

                          bool schedule_only = false);

  // Schedule a background job to actually delete obsolete files.

  void SchedulePurge();

  const SnapshotList& snapshots() const { return snapshots_; }

  // load list of snapshots to `snap_vector` that is no newer than `max_seq`

  // in ascending order.

  // `oldest_write_conflict_snapshot` is filled with the oldest snapshot

  // which satisfies SnapshotImpl.is_write_conflict_boundary_ = true.

  void LoadSnapshots(std::vector<SequenceNumber>* snap_vector,

                     SequenceNumber* oldest_write_conflict_snapshot,

                     const SequenceNumber& max_seq) const {

    InstrumentedMutexLock l(mutex());

    snapshots().GetAll(snap_vector, oldest_write_conflict_snapshot, max_seq);

  }

  const ImmutableDBOptions& immutable_db_options() const {

    return immutable_db_options_;

  }

  // Cancel all background jobs, including flush, compaction, background

  // purging, stats dumping threads, etc. If `wait` = true, wait for the

  // running jobs to abort or finish before returning. Otherwise, only

  // sends the signals.

  void CancelAllBackgroundWork(bool wait);

  // Find Super version and reference it. Based on options, it might return

  // the thread local cached one.

  // Call ReturnAndCleanupSuperVersion() when it is no longer needed.

  SuperVersion* GetAndRefSuperVersion(ColumnFamilyData* cfd);

  // Similar to the previous function but looks up based on a column family id.

  // nullptr will be returned if this column family no longer exists.

  // REQUIRED: this function should only be called on the write thread or if the

  // mutex is held.

  SuperVersion* GetAndRefSuperVersion(uint32_t column_family_id);

  // Un-reference the super version and clean it up if it is the last reference.

  void CleanupSuperVersion(SuperVersion* sv);

  // Un-reference the super version and return it to thread local cache if

  // needed. If it is the last reference of the super version. Clean it up

  // after un-referencing it.

  void ReturnAndCleanupSuperVersion(ColumnFamilyData* cfd, SuperVersion* sv);

  // Similar to the previous function but looks up based on a column family id.

  // nullptr will be returned if this column family no longer exists.

  // REQUIRED: this function should only be called on the write thread.

  void ReturnAndCleanupSuperVersion(uint32_t colun_family_id, SuperVersion* sv);

  // REQUIRED: this function should only be called on the write thread or if the

  // mutex is held.  Return value only valid until next call to this function or

  // mutex is released.

  ColumnFamilyHandle* GetColumnFamilyHandle(uint32_t column_family_id);

  // Same as above, should called without mutex held and not on write thread.

  std::unique_ptr<ColumnFamilyHandle> GetColumnFamilyHandleUnlocked(

      uint32_t column_family_id);

  // Returns the number of currently running flushes.

  // REQUIREMENT: mutex_ must be held when calling this function.

  int num_running_flushes() {

    mutex_.AssertHeld();

    return num_running_flushes_;

  }

  // Returns the number of currently running compactions.

  // REQUIREMENT: mutex_ must be held when calling this function.

  int num_running_compactions() {

    mutex_.AssertHeld();

    return num_running_compactions_;

  }

  const WriteController& write_controller() { return write_controller_; }

  // hollow transactions shell used for recovery.

  // these will then be passed to TransactionDB so that

  // locks can be reacquired before writing can resume.

  struct RecoveredTransaction {

    std::string name_;

    bool unprepared_;

    struct BatchInfo {

      uint64_t log_number_;

      // TODO(lth): For unprepared, the memory usage here can be big for

      // unprepared transactions. This is only useful for rollbacks, and we

      // can in theory just keep keyset for that.

      WriteBatch* batch_;

      // Number of sub-batches. A new sub-batch is created if txn attempts to

      // insert a duplicate key,seq to memtable. This is currently used in

      // WritePreparedTxn/WriteUnpreparedTxn.

      size_t batch_cnt_;

    };

    // This maps the seq of the first key in the batch to BatchInfo, which

    // contains WriteBatch and other information relevant to the batch.

    //

    // For WriteUnprepared, batches_ can have size greater than 1, but for

    // other write policies, it must be of size 1.

    std::map<SequenceNumber, BatchInfo> batches_;

    explicit RecoveredTransaction(const uint64_t log, const std::string& name,

                                  WriteBatch* batch, SequenceNumber seq,

                                  size_t batch_cnt, bool unprepared)

        : name_(name), unprepared_(unprepared) {

      batches_[seq] = {log, batch, batch_cnt};

    }

    ~RecoveredTransaction() {

      for (auto& it : batches_) {

        delete it.second.batch_;

      }

    }

    void AddBatch(SequenceNumber seq, uint64_t log_number, WriteBatch* batch,

                  size_t batch_cnt, bool unprepared) {

      assert(batches_.count(seq) == 0);

      batches_[seq] = {log_number, batch, batch_cnt};

      // Prior state must be unprepared, since the prepare batch must be the

      // last batch.

      assert(unprepared_);

      unprepared_ = unprepared;

    }

  };

  bool allow_2pc() const { return immutable_db_options_.allow_2pc; }

  std::unordered_map<std::string, RecoveredTransaction*>

  recovered_transactions() {

    return recovered_transactions_;

  }

  RecoveredTransaction* GetRecoveredTransaction(const std::string& name) {

    auto it = recovered_transactions_.find(name);

    if (it == recovered_transactions_.end()) {

      return nullptr;

    } else {

      return it->second;

    }

  }

  void InsertRecoveredTransaction(const uint64_t log, const std::string& name,

                                  WriteBatch* batch, SequenceNumber seq,

                                  size_t batch_cnt, bool unprepared_batch) {

    // For WriteUnpreparedTxn, InsertRecoveredTransaction is called multiple

    // times for every unprepared batch encountered during recovery.

    //

    // If the transaction is prepared, then the last call to

    // InsertRecoveredTransaction will have unprepared_batch = false.

    auto rtxn = recovered_transactions_.find(name);

    if (rtxn == recovered_transactions_.end()) {

      recovered_transactions_[name] = new RecoveredTransaction(

          log, name, batch, seq, batch_cnt, unprepared_batch);

    } else {

      rtxn->second->AddBatch(seq, log, batch, batch_cnt, unprepared_batch);

    }

    logs_with_prep_tracker_.MarkLogAsContainingPrepSection(log);

  }

  void DeleteRecoveredTransaction(const std::string& name) {

    auto it = recovered_transactions_.find(name);

    assert(it != recovered_transactions_.end());

    auto* trx = it->second;

    recovered_transactions_.erase(it);

    for (const auto& info : trx->batches_) {

      logs_with_prep_tracker_.MarkLogAsHavingPrepSectionFlushed(

          info.second.log_number_);

    }

    delete trx;

  }

  void DeleteAllRecoveredTransactions() {

    for (auto it = recovered_transactions_.begin();

         it != recovered_transactions_.end(); ++it) {

      delete it->second;

    }

    recovered_transactions_.clear();

  }

  void AddToLogsToFreeQueue(log::Writer* log_writer) {

    mutex_.AssertHeld();

    logs_to_free_queue_.push_back(log_writer);

  }

  void AddSuperVersionsToFreeQueue(SuperVersion* sv) {

    superversions_to_free_queue_.push_back(sv);

  }

  void SetSnapshotChecker(SnapshotChecker* snapshot_checker);

  // Fill JobContext with snapshot information needed by flush and compaction.

  void GetSnapshotContext(JobContext* job_context,

                          std::vector<SequenceNumber>* snapshot_seqs,

                          SequenceNumber* earliest_write_conflict_snapshot,

                          SnapshotChecker** snapshot_checker);

  // Not thread-safe.

  void SetRecoverableStatePreReleaseCallback(PreReleaseCallback* callback);

  InstrumentedMutex* mutex() const { return &mutex_; }

  // Initialize a brand new DB. The DB directory is expected to be empty before

  // calling it. Push new manifest file name into `new_filenames`.

  Status NewDB(std::vector<std::string>* new_filenames);

  // This is to be used only by internal rocksdb classes.

  static Status Open(const DBOptions& db_options, const std::string& name,

                     const std::vector<ColumnFamilyDescriptor>& column_families,

                     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr,

                     const bool seq_per_batch, const bool batch_per_txn,

                     const bool is_retry, bool* can_retry);

  static IOStatus CreateAndNewDirectory(

      FileSystem* fs, const std::string& dirname,

      std::unique_ptr<FSDirectory>* directory);

  // find stats map from stats_history_ with smallest timestamp in

  // the range of [start_time, end_time)

  bool FindStatsByTime(uint64_t start_time, uint64_t end_time,