//  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).
#include "options/options_helper.h"

#include <cassert>
#include <cctype>
#include <cstdlib>
#include <set>
#include <unordered_set>
#include <vector>

#include "options/cf_options.h"
#include "options/db_options.h"
#include "rocksdb/cache.h"
#include "rocksdb/compaction_filter.h"
#include "rocksdb/convenience.h"
#include "rocksdb/filter_policy.h"
#include "rocksdb/flush_block_policy.h"
#include "rocksdb/memtablerep.h"
#include "rocksdb/merge_operator.h"
#include "rocksdb/options.h"
#include "rocksdb/rate_limiter.h"
#include "rocksdb/slice_transform.h"
#include "rocksdb/table.h"
#include "rocksdb/utilities/object_registry.h"
#include "rocksdb/utilities/options_type.h"
#include "util/string_util.h"

namespace ROCKSDB_NAMESPACE {
ConfigOptions::ConfigOptions()
    : registry(ObjectRegistry::NewInstance())
{
  env = Env::Default();
}

ConfigOptions::ConfigOptions(const DBOptions& db_opts) : env(db_opts.env) {
  registry = ObjectRegistry::NewInstance();
}

Status ValidateOptions(const DBOptions& db_opts,
                       const ColumnFamilyOptions& cf_opts) {
  Status s;
  auto db_cfg = DBOptionsAsConfigurable(db_opts);
  auto cf_cfg = CFOptionsAsConfigurable(cf_opts);
  s = db_cfg->ValidateOptions(db_opts, cf_opts);
  if (s.ok()) s = cf_cfg->ValidateOptions(db_opts, cf_opts);
  return s;
}

DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
                         const MutableDBOptions& mutable_db_options) {
  DBOptions options;

  options.create_if_missing = immutable_db_options.create_if_missing;
  options.create_missing_column_families =
      immutable_db_options.create_missing_column_families;
  options.error_if_exists = immutable_db_options.error_if_exists;
  options.paranoid_checks = immutable_db_options.paranoid_checks;
  options.flush_verify_memtable_count =
      immutable_db_options.flush_verify_memtable_count;
  options.compaction_verify_record_count =
      immutable_db_options.compaction_verify_record_count;
  options.track_and_verify_wals_in_manifest =
      immutable_db_options.track_and_verify_wals_in_manifest;
  options.verify_sst_unique_id_in_manifest =
      immutable_db_options.verify_sst_unique_id_in_manifest;
  options.env = immutable_db_options.env;
  options.rate_limiter = immutable_db_options.rate_limiter;
  options.sst_file_manager = immutable_db_options.sst_file_manager;
  options.info_log = immutable_db_options.info_log;
  options.info_log_level = immutable_db_options.info_log_level;
  options.max_open_files = mutable_db_options.max_open_files;
  options.max_file_opening_threads =
      immutable_db_options.max_file_opening_threads;
  options.max_total_wal_size = mutable_db_options.max_total_wal_size;
  options.statistics = immutable_db_options.statistics;
  options.use_fsync = immutable_db_options.use_fsync;
  options.db_paths = immutable_db_options.db_paths;
  options.db_log_dir = immutable_db_options.db_log_dir;
  options.wal_dir = immutable_db_options.wal_dir;
  options.delete_obsolete_files_period_micros =
      mutable_db_options.delete_obsolete_files_period_micros;
  options.max_background_jobs = mutable_db_options.max_background_jobs;
  options.max_background_compactions =
      mutable_db_options.max_background_compactions;
  options.bytes_per_sync = mutable_db_options.bytes_per_sync;
  options.wal_bytes_per_sync = mutable_db_options.wal_bytes_per_sync;
  options.strict_bytes_per_sync = mutable_db_options.strict_bytes_per_sync;
  options.max_subcompactions = mutable_db_options.max_subcompactions;
  options.max_background_flushes = mutable_db_options.max_background_flushes;
  options.max_log_file_size = immutable_db_options.max_log_file_size;
  options.log_file_time_to_roll = immutable_db_options.log_file_time_to_roll;
  options.keep_log_file_num = immutable_db_options.keep_log_file_num;
  options.recycle_log_file_num = immutable_db_options.recycle_log_file_num;
  options.max_manifest_file_size = immutable_db_options.max_manifest_file_size;
  options.table_cache_numshardbits =
      immutable_db_options.table_cache_numshardbits;
  options.WAL_ttl_seconds = immutable_db_options.WAL_ttl_seconds;
  options.WAL_size_limit_MB = immutable_db_options.WAL_size_limit_MB;
  options.manifest_preallocation_size =
      immutable_db_options.manifest_preallocation_size;
  options.allow_mmap_reads = immutable_db_options.allow_mmap_reads;
  options.allow_mmap_writes = immutable_db_options.allow_mmap_writes;
  options.use_direct_reads = immutable_db_options.use_direct_reads;
  options.use_direct_io_for_flush_and_compaction =
      immutable_db_options.use_direct_io_for_flush_and_compaction;
  options.allow_fallocate = immutable_db_options.allow_fallocate;
  options.is_fd_close_on_exec = immutable_db_options.is_fd_close_on_exec;
  options.stats_dump_period_sec = mutable_db_options.stats_dump_period_sec;
  options.stats_persist_period_sec =
      mutable_db_options.stats_persist_period_sec;
  options.persist_stats_to_disk = immutable_db_options.persist_stats_to_disk;
  options.stats_history_buffer_size =
      mutable_db_options.stats_history_buffer_size;
  options.advise_random_on_open = immutable_db_options.advise_random_on_open;
  options.db_write_buffer_size = immutable_db_options.db_write_buffer_size;
  options.write_buffer_manager = immutable_db_options.write_buffer_manager;
  options.access_hint_on_compaction_start =
      immutable_db_options.access_hint_on_compaction_start;
  options.compaction_readahead_size =
      mutable_db_options.compaction_readahead_size;
  options.random_access_max_buffer_size =
      immutable_db_options.random_access_max_buffer_size;
  options.writable_file_max_buffer_size =
      mutable_db_options.writable_file_max_buffer_size;
  options.use_adaptive_mutex = immutable_db_options.use_adaptive_mutex;
  options.listeners = immutable_db_options.listeners;
  options.enable_thread_tracking = immutable_db_options.enable_thread_tracking;
  options.delayed_write_rate = mutable_db_options.delayed_write_rate;
  options.enable_pipelined_write = immutable_db_options.enable_pipelined_write;
  options.unordered_write = immutable_db_options.unordered_write;
  options.allow_concurrent_memtable_write =
      immutable_db_options.allow_concurrent_memtable_write;
  options.enable_write_thread_adaptive_yield =
      immutable_db_options.enable_write_thread_adaptive_yield;
  options.max_write_batch_group_size_bytes =
      immutable_db_options.max_write_batch_group_size_bytes;
  options.write_thread_max_yield_usec =
      immutable_db_options.write_thread_max_yield_usec;
  options.write_thread_slow_yield_usec =
      immutable_db_options.write_thread_slow_yield_usec;
  options.skip_stats_update_on_db_open =
      immutable_db_options.skip_stats_update_on_db_open;
  options.skip_checking_sst_file_sizes_on_db_open =
      immutable_db_options.skip_checking_sst_file_sizes_on_db_open;
  options.wal_recovery_mode = immutable_db_options.wal_recovery_mode;
  options.allow_2pc = immutable_db_options.allow_2pc;
  options.row_cache = immutable_db_options.row_cache;
  options.wal_filter = immutable_db_options.wal_filter;
  options.fail_if_options_file_error =
      immutable_db_options.fail_if_options_file_error;
  options.dump_malloc_stats = immutable_db_options.dump_malloc_stats;
  options.avoid_flush_during_recovery =
      immutable_db_options.avoid_flush_during_recovery;
  options.avoid_flush_during_shutdown =
      mutable_db_options.avoid_flush_during_shutdown;
  options.allow_ingest_behind = immutable_db_options.allow_ingest_behind;
  options.two_write_queues = immutable_db_options.two_write_queues;
  options.manual_wal_flush = immutable_db_options.manual_wal_flush;
  options.wal_compression = immutable_db_options.wal_compression;
  options.atomic_flush = immutable_db_options.atomic_flush;
  options.avoid_unnecessary_blocking_io =
      immutable_db_options.avoid_unnecessary_blocking_io;
  options.log_readahead_size = immutable_db_options.log_readahead_size;
  options.file_checksum_gen_factory =
      immutable_db_options.file_checksum_gen_factory;
  options.best_efforts_recovery = immutable_db_options.best_efforts_recovery;
  options.max_bgerror_resume_count =
      immutable_db_options.max_bgerror_resume_count;
  options.bgerror_resume_retry_interval =
      immutable_db_options.bgerror_resume_retry_interval;
  options.db_host_id = immutable_db_options.db_host_id;
  options.allow_data_in_errors = immutable_db_options.allow_data_in_errors;
  options.checksum_handoff_file_types =
      immutable_db_options.checksum_handoff_file_types;
  options.lowest_used_cache_tier = immutable_db_options.lowest_used_cache_tier;
  options.enforce_single_del_contracts =
      immutable_db_options.enforce_single_del_contracts;
  return options;
}

ColumnFamilyOptions BuildColumnFamilyOptions(
    const ColumnFamilyOptions& options,
    const MutableCFOptions& mutable_cf_options) {
  ColumnFamilyOptions cf_opts(options);
  UpdateColumnFamilyOptions(mutable_cf_options, &cf_opts);
  // TODO(yhchiang): find some way to handle the following derived options
  // * max_file_size
  return cf_opts;
}

void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
                               ColumnFamilyOptions* cf_opts) {
  // Memtable related options
  cf_opts->write_buffer_size = moptions.write_buffer_size;
  cf_opts->max_write_buffer_number = moptions.max_write_buffer_number;
  cf_opts->arena_block_size = moptions.arena_block_size;
  cf_opts->memtable_prefix_bloom_size_ratio =
      moptions.memtable_prefix_bloom_size_ratio;
  cf_opts->memtable_whole_key_filtering = moptions.memtable_whole_key_filtering;
  cf_opts->memtable_huge_page_size = moptions.memtable_huge_page_size;
  cf_opts->max_successive_merges = moptions.max_successive_merges;
  cf_opts->inplace_update_num_locks = moptions.inplace_update_num_locks;
  cf_opts->prefix_extractor = moptions.prefix_extractor;
  cf_opts->experimental_mempurge_threshold =
      moptions.experimental_mempurge_threshold;
  cf_opts->memtable_protection_bytes_per_key =
      moptions.memtable_protection_bytes_per_key;
  cf_opts->block_protection_bytes_per_key =
      moptions.block_protection_bytes_per_key;
  cf_opts->bottommost_file_compaction_delay =
      moptions.bottommost_file_compaction_delay;

  // Compaction related options
  cf_opts->disable_auto_compactions = moptions.disable_auto_compactions;
  cf_opts->soft_pending_compaction_bytes_limit =
      moptions.soft_pending_compaction_bytes_limit;
  cf_opts->hard_pending_compaction_bytes_limit =
      moptions.hard_pending_compaction_bytes_limit;
  cf_opts->level0_file_num_compaction_trigger =
      moptions.level0_file_num_compaction_trigger;
  cf_opts->level0_slowdown_writes_trigger =
      moptions.level0_slowdown_writes_trigger;
  cf_opts->level0_stop_writes_trigger = moptions.level0_stop_writes_trigger;
  cf_opts->max_compaction_bytes = moptions.max_compaction_bytes;
  cf_opts->ignore_max_compaction_bytes_for_input =
      moptions.ignore_max_compaction_bytes_for_input;
  cf_opts->target_file_size_base = moptions.target_file_size_base;
  cf_opts->target_file_size_multiplier = moptions.target_file_size_multiplier;
  cf_opts->max_bytes_for_level_base = moptions.max_bytes_for_level_base;
  cf_opts->max_bytes_for_level_multiplier =
      moptions.max_bytes_for_level_multiplier;
  cf_opts->ttl = moptions.ttl;
  cf_opts->periodic_compaction_seconds = moptions.periodic_compaction_seconds;

  cf_opts->max_bytes_for_level_multiplier_additional.clear();
  for (auto value : moptions.max_bytes_for_level_multiplier_additional) {
    cf_opts->max_bytes_for_level_multiplier_additional.emplace_back(value);
  }

  cf_opts->compaction_options_fifo = moptions.compaction_options_fifo;
  cf_opts->compaction_options_universal = moptions.compaction_options_universal;

  // Blob file related options
  cf_opts->enable_blob_files = moptions.enable_blob_files;
  cf_opts->min_blob_size = moptions.min_blob_size;
  cf_opts->blob_file_size = moptions.blob_file_size;
  cf_opts->blob_compression_type = moptions.blob_compression_type;
  cf_opts->enable_blob_garbage_collection =
      moptions.enable_blob_garbage_collection;
  cf_opts->blob_garbage_collection_age_cutoff =
      moptions.blob_garbage_collection_age_cutoff;
  cf_opts->blob_garbage_collection_force_threshold =
      moptions.blob_garbage_collection_force_threshold;
  cf_opts->blob_compaction_readahead_size =
      moptions.blob_compaction_readahead_size;
  cf_opts->blob_file_starting_level = moptions.blob_file_starting_level;
  cf_opts->prepopulate_blob_cache = moptions.prepopulate_blob_cache;

  // Misc options
  cf_opts->max_sequential_skip_in_iterations =
      moptions.max_sequential_skip_in_iterations;
  cf_opts->check_flush_compaction_key_order =
      moptions.check_flush_compaction_key_order;
  cf_opts->paranoid_file_checks = moptions.paranoid_file_checks;
  cf_opts->report_bg_io_stats = moptions.report_bg_io_stats;
  cf_opts->compression = moptions.compression;
  cf_opts->compression_opts = moptions.compression_opts;
  cf_opts->bottommost_compression = moptions.bottommost_compression;
  cf_opts->bottommost_compression_opts = moptions.bottommost_compression_opts;
  cf_opts->sample_for_compression = moptions.sample_for_compression;
  cf_opts->compression_per_level = moptions.compression_per_level;
  cf_opts->last_level_temperature = moptions.last_level_temperature;
  cf_opts->bottommost_temperature = moptions.last_level_temperature;
  cf_opts->memtable_max_range_deletions = moptions.memtable_max_range_deletions;
}

void UpdateColumnFamilyOptions(const ImmutableCFOptions& ioptions,
                               ColumnFamilyOptions* cf_opts) {
  cf_opts->compaction_style = ioptions.compaction_style;
  cf_opts->compaction_pri = ioptions.compaction_pri;
  cf_opts->comparator = ioptions.user_comparator;
  cf_opts->merge_operator = ioptions.merge_operator;
  cf_opts->compaction_filter = ioptions.compaction_filter;
  cf_opts->compaction_filter_factory = ioptions.compaction_filter_factory;
  cf_opts->min_write_buffer_number_to_merge =
      ioptions.min_write_buffer_number_to_merge;
  cf_opts->max_write_buffer_number_to_maintain =
      ioptions.max_write_buffer_number_to_maintain;
  cf_opts->max_write_buffer_size_to_maintain =
      ioptions.max_write_buffer_size_to_maintain;
  cf_opts->inplace_update_support = ioptions.inplace_update_support;
  cf_opts->inplace_callback = ioptions.inplace_callback;
  cf_opts->memtable_factory = ioptions.memtable_factory;
  cf_opts->table_factory = ioptions.table_factory;
  cf_opts->table_properties_collector_factories =
      ioptions.table_properties_collector_factories;
  cf_opts->bloom_locality = ioptions.bloom_locality;
  cf_opts->level_compaction_dynamic_level_bytes =
      ioptions.level_compaction_dynamic_level_bytes;
  cf_opts->level_compaction_dynamic_file_size =
      ioptions.level_compaction_dynamic_file_size;
  cf_opts->num_levels = ioptions.num_levels;
  cf_opts->optimize_filters_for_hits = ioptions.optimize_filters_for_hits;
  cf_opts->force_consistency_checks = ioptions.force_consistency_checks;
  cf_opts->memtable_insert_with_hint_prefix_extractor =
      ioptions.memtable_insert_with_hint_prefix_extractor;
  cf_opts->cf_paths = ioptions.cf_paths;
  cf_opts->compaction_thread_limiter = ioptions.compaction_thread_limiter;
  cf_opts->sst_partitioner_factory = ioptions.sst_partitioner_factory;
  cf_opts->blob_cache = ioptions.blob_cache;
  cf_opts->preclude_last_level_data_seconds =
      ioptions.preclude_last_level_data_seconds;
  cf_opts->preserve_internal_time_seconds =
      ioptions.preserve_internal_time_seconds;
  cf_opts->persist_user_defined_timestamps =
      ioptions.persist_user_defined_timestamps;
  cf_opts->default_temperature = ioptions.default_temperature;

  // TODO(yhchiang): find some way to handle the following derived options
  // * max_file_size
}

std::map<CompactionStyle, std::string>
    OptionsHelper::compaction_style_to_string = {
        {kCompactionStyleLevel, "kCompactionStyleLevel"},
        {kCompactionStyleUniversal, "kCompactionStyleUniversal"},
        {kCompactionStyleFIFO, "kCompactionStyleFIFO"},
        {kCompactionStyleNone, "kCompactionStyleNone"}};

std::map<CompactionPri, std::string> OptionsHelper::compaction_pri_to_string = {
    {kByCompensatedSize, "kByCompensatedSize"},
    {kOldestLargestSeqFirst, "kOldestLargestSeqFirst"},
    {kOldestSmallestSeqFirst, "kOldestSmallestSeqFirst"},
    {kMinOverlappingRatio, "kMinOverlappingRatio"},
    {kRoundRobin, "kRoundRobin"}};

std::map<CompactionStopStyle, std::string>
    OptionsHelper::compaction_stop_style_to_string = {
        {kCompactionStopStyleSimilarSize, "kCompactionStopStyleSimilarSize"},
        {kCompactionStopStyleTotalSize, "kCompactionStopStyleTotalSize"}};

std::map<Temperature, std::string> OptionsHelper::temperature_to_string = {
    {Temperature::kUnknown, "kUnknown"},
    {Temperature::kHot, "kHot"},
    {Temperature::kWarm, "kWarm"},
    {Temperature::kCold, "kCold"}};

std::unordered_map<std::string, ChecksumType>
    OptionsHelper::checksum_type_string_map = {{"kNoChecksum", kNoChecksum},
                                               {"kCRC32c", kCRC32c},
                                               {"kxxHash", kxxHash},
                                               {"kxxHash64", kxxHash64},
                                               {"kXXH3", kXXH3}};

std::unordered_map<std::string, CompressionType>
    OptionsHelper::compression_type_string_map = {
        {"kNoCompression", kNoCompression},
        {"kSnappyCompression", kSnappyCompression},
        {"kZlibCompression", kZlibCompression},
        {"kBZip2Compression", kBZip2Compression},
        {"kLZ4Compression", kLZ4Compression},
        {"kLZ4HCCompression", kLZ4HCCompression},
        {"kXpressCompression", kXpressCompression},
        {"kZSTD", kZSTD},
        {"kZSTDNotFinalCompression", kZSTDNotFinalCompression},
        {"kDisableCompressionOption", kDisableCompressionOption}};

std::vector<CompressionType> GetSupportedCompressions() {
  // std::set internally to deduplicate potential name aliases
  std::set<CompressionType> supported_compressions;
  for (const auto& comp_to_name : OptionsHelper::compression_type_string_map) {
    CompressionType t = comp_to_name.second;
    if (t != kDisableCompressionOption && CompressionTypeSupported(t)) {
      supported_compressions.insert(t);
    }
  }
  return std::vector<CompressionType>(supported_compressions.begin(),
                                      supported_compressions.end());
}

std::vector<CompressionType> GetSupportedDictCompressions() {
  std::set<CompressionType> dict_compression_types;
  for (const auto& comp_to_name : OptionsHelper::compression_type_string_map) {
    CompressionType t = comp_to_name.second;
    if (t != kDisableCompressionOption && DictCompressionTypeSupported(t)) {
      dict_compression_types.insert(t);
    }
  }
  return std::vector<CompressionType>(dict_compression_types.begin(),
                                      dict_compression_types.end());
}

std::vector<ChecksumType> GetSupportedChecksums() {
  std::set<ChecksumType> checksum_types;
  for (const auto& e : OptionsHelper::checksum_type_string_map) {
    checksum_types.insert(e.second);
  }
  return std::vector<ChecksumType>(checksum_types.begin(),
                                   checksum_types.end());
}

static bool ParseOptionHelper(void* opt_address, const OptionType& opt_type,
                              const std::string& value) {
  switch (opt_type) {
    case OptionType::kBoolean:
      *static_cast<bool*>(opt_address) = ParseBoolean("", value);
      break;
    case OptionType::kInt:
      *static_cast<int*>(opt_address) = ParseInt(value);
      break;
    case OptionType::kInt32T:
      *static_cast<int32_t*>(opt_address) = ParseInt32(value);
      break;
    case OptionType::kInt64T:
      PutUnaligned(static_cast<int64_t*>(opt_address), ParseInt64(value));
      break;
    case OptionType::kUInt:
      *static_cast<unsigned int*>(opt_address) = ParseUint32(value);
      break;
    case OptionType::kUInt8T:
      *static_cast<uint8_t*>(opt_address) = ParseUint8(value);
      break;
    case OptionType::kUInt32T:
      *static_cast<uint32_t*>(opt_address) = ParseUint32(value);
      break;
    case OptionType::kUInt64T:
      PutUnaligned(static_cast<uint64_t*>(opt_address), ParseUint64(value));
      break;
    case OptionType::kSizeT:
      PutUnaligned(static_cast<size_t*>(opt_address), ParseSizeT(value));
      break;
    case OptionType::kString:
      *static_cast<std::string*>(opt_address) = value;
      break;
    case OptionType::kDouble:
      *static_cast<double*>(opt_address) = ParseDouble(value);
      break;
    case OptionType::kCompactionStyle:
      return ParseEnum<CompactionStyle>(
          compaction_style_string_map, value,
          static_cast<CompactionStyle*>(opt_address));
    case OptionType::kCompactionPri:
      return ParseEnum<CompactionPri>(compaction_pri_string_map, value,
                                      static_cast<CompactionPri*>(opt_address));
    case OptionType::kCompressionType:
      return ParseEnum<CompressionType>(
          compression_type_string_map, value,
          static_cast<CompressionType*>(opt_address));
    case OptionType::kChecksumType:
      return ParseEnum<ChecksumType>(checksum_type_string_map, value,
                                     static_cast<ChecksumType*>(opt_address));
    case OptionType::kEncodingType:
      return ParseEnum<EncodingType>(encoding_type_string_map, value,
                                     static_cast<EncodingType*>(opt_address));
    case OptionType::kCompactionStopStyle:
      return ParseEnum<CompactionStopStyle>(
          compaction_stop_style_string_map, value,
          static_cast<CompactionStopStyle*>(opt_address));
    case OptionType::kEncodedString: {
      std::string* output_addr = static_cast<std::string*>(opt_address);
      (Slice(value)).DecodeHex(output_addr);
      break;
    }
    case OptionType::kTemperature: {
      return ParseEnum<Temperature>(temperature_string_map, value,
                                    static_cast<Temperature*>(opt_address));
    }
    default:
      return false;
  }
  return true;
}

bool SerializeSingleOptionHelper(const void* opt_address,
                                 const OptionType opt_type,
                                 std::string* value) {
  assert(value);
  switch (opt_type) {
    case OptionType::kBoolean:
      *value = *(static_cast<const bool*>(opt_address)) ? "true" : "false";
      break;
    case OptionType::kInt:
      *value = std::to_string(*(static_cast<const int*>(opt_address)));
      break;
    case OptionType::kInt32T:
      *value = std::to_string(*(static_cast<const int32_t*>(opt_address)));
      break;
    case OptionType::kInt64T:
      {
        int64_t v;
        GetUnaligned(static_cast<const int64_t*>(opt_address), &v);
        *value = std::to_string(v);
      }
      break;
    case OptionType::kUInt:
      *value = std::to_string(*(static_cast<const unsigned int*>(opt_address)));
      break;
    case OptionType::kUInt8T:
      *value = std::to_string(*(static_cast<const uint8_t*>(opt_address)));
      break;
    case OptionType::kUInt32T:
      *value = std::to_string(*(static_cast<const uint32_t*>(opt_address)));
      break;
    case OptionType::kUInt64T:
      {
        uint64_t v;
        GetUnaligned(static_cast<const uint64_t*>(opt_address), &v);
        *value = std::to_string(v);
      }
      break;
    case OptionType::kSizeT:
      {
        size_t v;
        GetUnaligned(static_cast<const size_t*>(opt_address), &v);
        *value = std::to_string(v);
      }
      break;
    case OptionType::kDouble:
      *value = std::to_string(*(static_cast<const double*>(opt_address)));
      break;
    case OptionType::kString:
      *value =
          EscapeOptionString(*(static_cast<const std::string*>(opt_address)));
      break;
    case OptionType::kCompactionStyle:
      return SerializeEnum<CompactionStyle>(
          compaction_style_string_map,
          *(static_cast<const CompactionStyle*>(opt_address)), value);
    case OptionType::kCompactionPri:
      return SerializeEnum<CompactionPri>(
          compaction_pri_string_map,
          *(static_cast<const CompactionPri*>(opt_address)), value);
    case OptionType::kCompressionType:
      return SerializeEnum<CompressionType>(
          compression_type_string_map,
          *(static_cast<const CompressionType*>(opt_address)), value);
      break;
    case OptionType::kChecksumType:
      return SerializeEnum<ChecksumType>(
          checksum_type_string_map,
          *static_cast<const ChecksumType*>(opt_address), value);
    case OptionType::kEncodingType:
      return SerializeEnum<EncodingType>(
          encoding_type_string_map,
          *static_cast<const EncodingType*>(opt_address), value);
    case OptionType::kCompactionStopStyle:
      return SerializeEnum<CompactionStopStyle>(
          compaction_stop_style_string_map,
          *static_cast<const CompactionStopStyle*>(opt_address), value);
    case OptionType::kEncodedString: {
      const auto* ptr = static_cast<const std::string*>(opt_address);
      *value = (Slice(*ptr)).ToString(true);
      break;
    }
    case OptionType::kTemperature: {
      return SerializeEnum<Temperature>(
          temperature_string_map, *static_cast<const Temperature*>(opt_address),
          value);
    }
    default:
      return false;
  }
  return true;
}

template <typename T>
Status ConfigureFromMap(
    const ConfigOptions& config_options,
    const std::unordered_map<std::string, std::string>& opt_map,
    const std::string& option_name, Configurable* config, T* new_opts) {
  Status s = config->ConfigureFromMap(config_options, opt_map);
  if (s.ok()) {
    *new_opts = *(config->GetOptions<T>(option_name));
  }
  return s;
}


Status StringToMap(const std::string& opts_str,
                   std::unordered_map<std::string, std::string>* opts_map) {
  assert(opts_map);
  // Example:
  //   opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
  //              "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
  size_t pos = 0;
  std::string opts = trim(opts_str);
  // If the input string starts and ends with "{...}", strip off the brackets
  while (opts.size() > 2 && opts[0] == '{' && opts[opts.size() - 1] == '}') {
    opts = trim(opts.substr(1, opts.size() - 2));
  }

  while (pos < opts.size()) {
    size_t eq_pos = opts.find_first_of("={};", pos);
    if (eq_pos == std::string::npos) {
      return Status::InvalidArgument("Mismatched key value pair, '=' expected");
    } else if (opts[eq_pos] != '=') {
      return Status::InvalidArgument("Unexpected char in key");
    }

    std::string key = trim(opts.substr(pos, eq_pos - pos));
    if (key.empty()) {
      return Status::InvalidArgument("Empty key found");
    }

    std::string value;
    Status s = OptionTypeInfo::NextToken(opts, ';', eq_pos + 1, &pos, &value);
    if (!s.ok()) {
      return s;
    } else {
      (*opts_map)[key] = value;
      if (pos == std::string::npos) {
        break;
      } else {
        pos++;
      }
    }
  }

  return Status::OK();
}


Status GetStringFromDBOptions(std::string* opt_string,
                              const DBOptions& db_options,
                              const std::string& delimiter) {
  ConfigOptions config_options(db_options);
  config_options.delimiter = delimiter;
  return GetStringFromDBOptions(config_options, db_options, opt_string);
}

Status GetStringFromDBOptions(const ConfigOptions& config_options,
                              const DBOptions& db_options,
                              std::string* opt_string) {
  assert(opt_string);
  opt_string->clear();
  auto config = DBOptionsAsConfigurable(db_options);
  return config->GetOptionString(config_options, opt_string);
}


Status GetStringFromColumnFamilyOptions(std::string* opt_string,
                                        const ColumnFamilyOptions& cf_options,
                                        const std::string& delimiter) {
  ConfigOptions config_options;
  config_options.delimiter = delimiter;
  return GetStringFromColumnFamilyOptions(config_options, cf_options,
                                          opt_string);
}

Status GetStringFromColumnFamilyOptions(const ConfigOptions& config_options,
                                        const ColumnFamilyOptions& cf_options,
                                        std::string* opt_string) {
  const auto config = CFOptionsAsConfigurable(cf_options);
  return config->GetOptionString(config_options, opt_string);
}

Status GetStringFromCompressionType(std::string* compression_str,
                                    CompressionType compression_type) {
  bool ok = SerializeEnum<CompressionType>(compression_type_string_map,
                                           compression_type, compression_str);
  if (ok) {
    return Status::OK();
  } else {
    return Status::InvalidArgument("Invalid compression types");
  }
}

Status GetColumnFamilyOptionsFromMap(
    const ConfigOptions& config_options,
    const ColumnFamilyOptions& base_options,
    const std::unordered_map<std::string, std::string>& opts_map,
    ColumnFamilyOptions* new_options) {
  assert(new_options);

  *new_options = base_options;

  const auto config = CFOptionsAsConfigurable(base_options);
  Status s = ConfigureFromMap<ColumnFamilyOptions>(
      config_options, opts_map, OptionsHelper::kCFOptionsName, config.get(),
      new_options);
  // Translate any errors (NotFound, NotSupported, to InvalidArgument
  if (s.ok() || s.IsInvalidArgument()) {
    return s;
  } else {
    return Status::InvalidArgument(s.getState());
  }
}

Status GetColumnFamilyOptionsFromString(const ConfigOptions& config_options,
                                        const ColumnFamilyOptions& base_options,
                                        const std::string& opts_str,
                                        ColumnFamilyOptions* new_options) {
  std::unordered_map<std::string, std::string> opts_map;
  Status s = StringToMap(opts_str, &opts_map);
  if (!s.ok()) {
    *new_options = base_options;
    return s;
  }
  return GetColumnFamilyOptionsFromMap(config_options, base_options, opts_map,
                                       new_options);
}

Status GetDBOptionsFromMap(
    const ConfigOptions& config_options, const DBOptions& base_options,
    const std::unordered_map<std::string, std::string>& opts_map,
    DBOptions* new_options) {
  assert(new_options);
  *new_options = base_options;
  auto config = DBOptionsAsConfigurable(base_options);
  Status s = ConfigureFromMap<DBOptions>(config_options, opts_map,
                                         OptionsHelper::kDBOptionsName,
                                         config.get(), new_options);
  // Translate any errors (NotFound, NotSupported, to InvalidArgument
  if (s.ok() || s.IsInvalidArgument()) {
    return s;
  } else {
    return Status::InvalidArgument(s.getState());
  }
}

Status GetDBOptionsFromString(const ConfigOptions& config_options,
                              const DBOptions& base_options,
                              const std::string& opts_str,
                              DBOptions* new_options) {
  std::unordered_map<std::string, std::string> opts_map;
  Status s = StringToMap(opts_str, &opts_map);
  if (!s.ok()) {
    *new_options = base_options;
    return s;
  }
  return GetDBOptionsFromMap(config_options, base_options, opts_map,
                             new_options);
}

Status GetOptionsFromString(const Options& base_options,
                            const std::string& opts_str, Options* new_options) {
  ConfigOptions config_options(base_options);
  config_options.input_strings_escaped = false;
  config_options.ignore_unknown_options = false;

  return GetOptionsFromString(config_options, base_options, opts_str,
                              new_options);
}

Status GetOptionsFromString(const ConfigOptions& config_options,
                            const Options& base_options,
                            const std::string& opts_str, Options* new_options) {
  ColumnFamilyOptions new_cf_options;
  std::unordered_map<std::string, std::string> unused_opts;
  std::unordered_map<std::string, std::string> opts_map;

  assert(new_options);
  *new_options = base_options;
  Status s = StringToMap(opts_str, &opts_map);
  if (!s.ok()) {
    return s;
  }
  auto config = DBOptionsAsConfigurable(base_options);
  s = config->ConfigureFromMap(config_options, opts_map, &unused_opts);

  if (s.ok()) {
    DBOptions* new_db_options =
        config->GetOptions<DBOptions>(OptionsHelper::kDBOptionsName);
    if (!unused_opts.empty()) {
      s = GetColumnFamilyOptionsFromMap(config_options, base_options,
                                        unused_opts, &new_cf_options);
      if (s.ok()) {
        *new_options = Options(*new_db_options, new_cf_options);
      }
    } else {
      *new_options = Options(*new_db_options, base_options);
    }
  }
  // Translate any errors (NotFound, NotSupported, to InvalidArgument
  if (s.ok() || s.IsInvalidArgument()) {
    return s;
  } else {
    return Status::InvalidArgument(s.getState());
  }
}

std::unordered_map<std::string, EncodingType>
    OptionsHelper::encoding_type_string_map = {{"kPlain", kPlain},
                                               {"kPrefix", kPrefix}};

std::unordered_map<std::string, CompactionStyle>
    OptionsHelper::compaction_style_string_map = {
        {"kCompactionStyleLevel", kCompactionStyleLevel},
        {"kCompactionStyleUniversal", kCompactionStyleUniversal},
        {"kCompactionStyleFIFO", kCompactionStyleFIFO},
        {"kCompactionStyleNone", kCompactionStyleNone}};

std::unordered_map<std::string, CompactionPri>
    OptionsHelper::compaction_pri_string_map = {
        {"kByCompensatedSize", kByCompensatedSize},
        {"kOldestLargestSeqFirst", kOldestLargestSeqFirst},
        {"kOldestSmallestSeqFirst", kOldestSmallestSeqFirst},
        {"kMinOverlappingRatio", kMinOverlappingRatio},
        {"kRoundRobin", kRoundRobin}};

std::unordered_map<std::string, CompactionStopStyle>
    OptionsHelper::compaction_stop_style_string_map = {
        {"kCompactionStopStyleSimilarSize", kCompactionStopStyleSimilarSize},
        {"kCompactionStopStyleTotalSize", kCompactionStopStyleTotalSize}};

std::unordered_map<std::string, Temperature>
    OptionsHelper::temperature_string_map = {
        {"kUnknown", Temperature::kUnknown},
        {"kHot", Temperature::kHot},
        {"kWarm", Temperature::kWarm},
        {"kCold", Temperature::kCold}};

std::unordered_map<std::string, PrepopulateBlobCache>
    OptionsHelper::prepopulate_blob_cache_string_map = {
        {"kDisable", PrepopulateBlobCache::kDisable},
        {"kFlushOnly", PrepopulateBlobCache::kFlushOnly}};

Status OptionTypeInfo::NextToken(const std::string& opts, char delimiter,
                                 size_t pos, size_t* end, std::string* token) {
  while (pos < opts.size() && isspace(opts[pos])) {
    ++pos;
  }
  // Empty value at the end
  if (pos >= opts.size()) {
    *token = "";
    *end = std::string::npos;
    return Status::OK();
  } else if (opts[pos] == '{') {
    int count = 1;
    size_t brace_pos = pos + 1;
    while (brace_pos < opts.size()) {
      if (opts[brace_pos] == '{') {
        ++count;
      } else if (opts[brace_pos] == '}') {
        --count;
        if (count == 0) {
          break;
        }
      }
      ++brace_pos;
    }
    // found the matching closing brace
    if (count == 0) {
      *token = trim(opts.substr(pos + 1, brace_pos - pos - 1));
      // skip all whitespace and move to the next delimiter
      // brace_pos points to the next position after the matching '}'
      pos = brace_pos + 1;
      while (pos < opts.size() && isspace(opts[pos])) {
        ++pos;
      }
      if (pos < opts.size() && opts[pos] != delimiter) {
        return Status::InvalidArgument("Unexpected chars after nested options");
      }
      *end = pos;
    } else {
      return Status::InvalidArgument(
          "Mismatched curly braces for nested options");
    }
  } else {
    *end = opts.find(delimiter, pos);
    if (*end == std::string::npos) {
      // It either ends with a trailing semi-colon or the last key-value pair
      *token = trim(opts.substr(pos));
    } else {
      *token = trim(opts.substr(pos, *end - pos));
    }
  }
  return Status::OK();
}

Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
                             const std::string& opt_name,
                             const std::string& value, void* opt_ptr) const {
  if (IsDeprecated()) {
    return Status::OK();
  }
  try {
    const std::string& opt_value = config_options.input_strings_escaped
                                       ? UnescapeOptionString(value)
                                       : value;

    if (opt_ptr == nullptr) {
      return Status::NotFound("Could not find option", opt_name);
    } else if (parse_func_ != nullptr) {
      ConfigOptions copy = config_options;
      copy.invoke_prepare_options = false;
      void* opt_addr = GetOffset(opt_ptr);
      return parse_func_(copy, opt_name, opt_value, opt_addr);
    } else if (ParseOptionHelper(GetOffset(opt_ptr), type_, opt_value)) {
      return Status::OK();
    } else if (IsConfigurable()) {
      // The option is <config>.<name>
      Configurable* config = AsRawPointer<Configurable>(opt_ptr);
      if (opt_value.empty()) {
        return Status::OK();
      } else if (config == nullptr) {
        return Status::NotFound("Could not find configurable: ", opt_name);
      } else {
        ConfigOptions copy = config_options;
        copy.ignore_unknown_options = false;
        copy.invoke_prepare_options = false;
        if (opt_value.find("=") != std::string::npos) {
          return config->ConfigureFromString(copy, opt_value);
        } else {
          return config->ConfigureOption(copy, opt_name, opt_value);
        }
      }
    } else if (IsByName()) {
      return Status::NotSupported("Deserializing the option " + opt_name +
                                  " is not supported");
    } else {
      return Status::InvalidArgument("Error parsing:", opt_name);
    }
  } catch (std::exception& e) {
    return Status::InvalidArgument("Error parsing " + opt_name + ":" +
                                   std::string(e.what()));
  }
}

Status OptionTypeInfo::ParseType(
    const ConfigOptions& config_options, const std::string& opts_str,
    const std::unordered_map<std::string, OptionTypeInfo>& type_map,
    void* opt_addr, std::unordered_map<std::string, std::string>* unused) {
  std::unordered_map<std::string, std::string> opts_map;
  Status status = StringToMap(opts_str, &opts_map);
  if (!status.ok()) {
    return status;
  } else {
    return ParseType(config_options, opts_map, type_map, opt_addr, unused);
  }
}

Status OptionTypeInfo::ParseType(
    const ConfigOptions& config_options,
    const std::unordered_map<std::string, std::string>& opts_map,
    const std::unordered_map<std::string, OptionTypeInfo>& type_map,
    void* opt_addr, std::unordered_map<std::string, std::string>* unused) {
  for (const auto& opts_iter : opts_map) {
    std::string opt_name;
    const auto* opt_info = Find(opts_iter.first, type_map, &opt_name);
    if (opt_info != nullptr) {
      Status status =
          opt_info->Parse(config_options, opt_name, opts_iter.second, opt_addr);
      if (!status.ok()) {
        return status;
      }
    } else if (unused != nullptr) {
      (*unused)[opts_iter.first] = opts_iter.second;
    } else if (!config_options.ignore_unknown_options) {
      return Status::NotFound("Unrecognized option", opts_iter.first);
    }
  }
  return Status::OK();
}

Status OptionTypeInfo::ParseStruct(
    const ConfigOptions& config_options, const std::string& struct_name,
    const std::unordered_map<std::string, OptionTypeInfo>* struct_map,
    const std::string& opt_name, const std::string& opt_value, void* opt_addr) {
  assert(struct_map);
  Status status;
  if (opt_name == struct_name || EndsWith(opt_name, "." + struct_name)) {
    // This option represents the entire struct
    std::unordered_map<std::string, std::string> unused;
    status =
        ParseType(config_options, opt_value, *struct_map, opt_addr, &unused);
    if (status.ok() && !unused.empty()) {
      status = Status::InvalidArgument(
          "Unrecognized option", struct_name + "." + unused.begin()->first);
    }
  } else if (StartsWith(opt_name, struct_name + ".")) {
    // This option represents a nested field in the struct (e.g, struct.field)
    std::string elem_name;
    const auto opt_info =
        Find(opt_name.substr(struct_name.size() + 1), *struct_map, &elem_name);
    if (opt_info != nullptr) {
      status = opt_info->Parse(config_options, elem_name, opt_value, opt_addr);
    } else {
      status = Status::InvalidArgument("Unrecognized option", opt_name);
    }
  } else {
    // This option represents a field in the struct (e.g. field)
    std::string elem_name;
    const auto opt_info = Find(opt_name, *struct_map, &elem_name);
    if (opt_info != nullptr) {
      status = opt_info->Parse(config_options, elem_name, opt_value, opt_addr);
    } else {
      status = Status::InvalidArgument("Unrecognized option",
                                       struct_name + "." + opt_name);
    }
  }
  return status;
}

Status OptionTypeInfo::Serialize(const ConfigOptions& config_options,
                                 const std::string& opt_name,
                                 const void* const opt_ptr,
                                 std::string* opt_value) const {
  // If the option is no longer used in rocksdb and marked as deprecated,
  // we skip it in the serialization.
  if (opt_ptr == nullptr || IsDeprecated()) {
    return Status::OK();
  } else if (IsEnabled(OptionTypeFlags::kDontSerialize)) {
    return Status::NotSupported("Cannot serialize option: ", opt_name);
  } else if (serialize_func_ != nullptr) {
    const void* opt_addr = GetOffset(opt_ptr);
    return serialize_func_(config_options, opt_name, opt_addr, opt_value);
  } else if (IsCustomizable()) {
    const Customizable* custom = AsRawPointer<Customizable>(opt_ptr);
    opt_value->clear();
    if (custom == nullptr) {
      // We do not have a custom object to serialize.
      // If the option is not mutable and we are doing only mutable options,
      // we return an empty string (which will cause the option not to be
      // printed). Otherwise, we return the "nullptr" string, which will result
      // in "option=nullptr" being printed.
      if (IsMutable() || !config_options.mutable_options_only) {
        *opt_value = kNullptrString;
      } else {
        *opt_value = "";
      }
    } else if (IsEnabled(OptionTypeFlags::kStringNameOnly) &&
               !config_options.IsDetailed()) {
      if (!config_options.mutable_options_only || IsMutable()) {
        *opt_value = custom->GetId();
      }
    } else {
      ConfigOptions embedded = config_options;
      embedded.delimiter = ";";
      // If this option is mutable, everything inside it should be considered
      // mutable
      if (IsMutable()) {
        embedded.mutable_options_only = false;
      }
      std::string value = custom->ToString(embedded);
      if (!embedded.mutable_options_only ||
          value.find("=") != std::string::npos) {
        *opt_value = value;
      } else {
        *opt_value = "";
      }
    }
    return Status::OK();
  } else if (IsConfigurable()) {
    const Configurable* config = AsRawPointer<Configurable>(opt_ptr);
    if (config != nullptr) {
      ConfigOptions embedded = config_options;
      embedded.delimiter = ";";
      *opt_value = config->ToString(embedded);
    }
    return Status::OK();
  } else if (config_options.mutable_options_only && !IsMutable()) {
    return Status::OK();
  } else if (SerializeSingleOptionHelper(GetOffset(opt_ptr), type_,
                                         opt_value)) {
    return Status::OK();
  } else {
    return Status::InvalidArgument("Cannot serialize option: ", opt_name);
  }
}

Status OptionTypeInfo::SerializeType(
    const ConfigOptions& config_options,
    const std::unordered_map<std::string, OptionTypeInfo>& type_map,
    const void* opt_addr, std::string* result) {
  Status status;
  for (const auto& iter : type_map) {
    std::string single;
    const auto& opt_info = iter.second;
    if (opt_info.ShouldSerialize()) {
      status =
          opt_info.Serialize(config_options, iter.first, opt_addr, &single);
      if (!status.ok()) {
        return status;
      } else {
        result->append(iter.first + "=" + single + config_options.delimiter);
      }
    }
  }
  return status;
}

Status OptionTypeInfo::SerializeStruct(
    const ConfigOptions& config_options, const std::string& struct_name,
    const std::unordered_map<std::string, OptionTypeInfo>* struct_map,
    const std::string& opt_name, const void* opt_addr, std::string* value) {
  assert(struct_map);
  Status status;
  if (EndsWith(opt_name, struct_name)) {
    // We are going to write the struct as "{ prop1=value1; prop2=value2;}.
    // Set the delimiter to ";" so that the everything will be on one line.
    ConfigOptions embedded = config_options;
    embedded.delimiter = ";";

    // This option represents the entire struct
    std::string result;
    status = SerializeType(embedded, *struct_map, opt_addr, &result);
    if (!status.ok()) {
      return status;
    } else {
      *value = "{" + result + "}";
    }
  } else if (StartsWith(opt_name, struct_name + ".")) {
    // This option represents a nested field in the struct (e.g, struct.field)
    std::string elem_name;
    const auto opt_info =
        Find(opt_name.substr(struct_name.size() + 1), *struct_map, &elem_name);
    if (opt_info != nullptr) {
      status = opt_info->Serialize(config_options, elem_name, opt_addr, value);
    } else {
      status = Status::InvalidArgument("Unrecognized option", opt_name);
    }
  } else {
    // This option represents a field in the struct (e.g. field)
    std::string elem_name;
    const auto opt_info = Find(opt_name, *struct_map, &elem_name);
    if (opt_info == nullptr) {
      status = Status::InvalidArgument("Unrecognized option", opt_name);
    } else if (opt_info->ShouldSerialize()) {
      status = opt_info->Serialize(config_options, opt_name + "." + elem_name,
                                   opt_addr, value);
    }
  }
  return status;
}

template <typename T>
bool IsOptionEqual(const void* offset1, const void* offset2) {
  return (*static_cast<const T*>(offset1) == *static_cast<const T*>(offset2));
}

static bool AreEqualDoubles(const double a, const double b) {
  return (fabs(a - b) < 0.00001);
}

static bool AreOptionsEqual(OptionType type, const void* this_offset,
                            const void* that_offset) {
  switch (type) {
    case OptionType::kBoolean:
      return IsOptionEqual<bool>(this_offset, that_offset);
    case OptionType::kInt:
      return IsOptionEqual<int>(this_offset, that_offset);
    case OptionType::kUInt:
      return IsOptionEqual<unsigned int>(this_offset, that_offset);
    case OptionType::kInt32T:
      return IsOptionEqual<int32_t>(this_offset, that_offset);
    case OptionType::kInt64T: {
      int64_t v1, v2;
      GetUnaligned(static_cast<const int64_t*>(this_offset), &v1);
      GetUnaligned(static_cast<const int64_t*>(that_offset), &v2);
      return (v1 == v2);
    }
    case OptionType::kUInt8T:
      return IsOptionEqual<uint8_t>(this_offset, that_offset);
    case OptionType::kUInt32T:
      return IsOptionEqual<uint32_t>(this_offset, that_offset);
    case OptionType::kUInt64T: {
      uint64_t v1, v2;
      GetUnaligned(static_cast<const uint64_t*>(this_offset), &v1);
      GetUnaligned(static_cast<const uint64_t*>(that_offset), &v2);
      return (v1 == v2);
    }
    case OptionType::kSizeT: {
      size_t v1, v2;
      GetUnaligned(static_cast<const size_t*>(this_offset), &v1);
      GetUnaligned(static_cast<const size_t*>(that_offset), &v2);
      return (v1 == v2);
    }
    case OptionType::kString:
      return IsOptionEqual<std::string>(this_offset, that_offset);
    case OptionType::kDouble:
      return AreEqualDoubles(*static_cast<const double*>(this_offset),
                             *static_cast<const double*>(that_offset));
    case OptionType::kCompactionStyle:
      return IsOptionEqual<CompactionStyle>(this_offset, that_offset);
    case OptionType::kCompactionStopStyle:
      return IsOptionEqual<CompactionStopStyle>(this_offset, that_offset);
    case OptionType::kCompactionPri:
      return IsOptionEqual<CompactionPri>(this_offset, that_offset);
    case OptionType::kCompressionType:
      return IsOptionEqual<CompressionType>(this_offset, that_offset);
    case OptionType::kChecksumType:
      return IsOptionEqual<ChecksumType>(this_offset, that_offset);
    case OptionType::kEncodingType:
      return IsOptionEqual<EncodingType>(this_offset, that_offset);
    case OptionType::kEncodedString:
      return IsOptionEqual<std::string>(this_offset, that_offset);
    case OptionType::kTemperature:
      return IsOptionEqual<Temperature>(this_offset, that_offset);
    default:
      return false;
  }  // End switch
}

bool OptionTypeInfo::AreEqual(const ConfigOptions& config_options,
                              const std::string& opt_name,
                              const void* const this_ptr,
                              const void* const that_ptr,
                              std::string* mismatch) const {
  auto level = GetSanityLevel();
  if (!config_options.IsCheckEnabled(level)) {
    return true;  // If the sanity level is not being checked, skip it
  }
  if (this_ptr == nullptr || that_ptr == nullptr) {
    if (this_ptr == that_ptr) {
      return true;
    }
  } else if (equals_func_ != nullptr) {
    const void* this_addr = GetOffset(this_ptr);
    const void* that_addr = GetOffset(that_ptr);
    if (equals_func_(config_options, opt_name, this_addr, that_addr,
                     mismatch)) {
      return true;
    }
  } else {
    const void* this_addr = GetOffset(this_ptr);
    const void* that_addr = GetOffset(that_ptr);
    if (AreOptionsEqual(type_, this_addr, that_addr)) {
      return true;
    } else if (IsConfigurable()) {
      const auto* this_config = AsRawPointer<Configurable>(this_ptr);
      const auto* that_config = AsRawPointer<Configurable>(that_ptr);
      if (this_config == that_config) {
        return true;
      } else if (this_config != nullptr && that_config != nullptr) {
        std::string bad_name;
        bool matches;
        if (level < config_options.sanity_level) {
          ConfigOptions copy = config_options;
          copy.sanity_level = level;
          matches = this_config->AreEquivalent(copy, that_config, &bad_name);
        } else {
          matches = this_config->AreEquivalent(config_options, that_config,
                                               &bad_name);
        }
        if (!matches) {
          *mismatch = opt_name + "." + bad_name;
        }
        return matches;
      }
    }
  }
  if (mismatch->empty()) {
    *mismatch = opt_name;
  }
  return false;
}

bool OptionTypeInfo::TypesAreEqual(
    const ConfigOptions& config_options,
    const std::unordered_map<std::string, OptionTypeInfo>& type_map,
    const void* this_addr, const void* that_addr, std::string* mismatch) {
  for (const auto& iter : type_map) {
    const auto& opt_info = iter.second;
    if (!opt_info.AreEqual(config_options, iter.first, this_addr, that_addr,
                           mismatch)) {
      return false;
    }
  }
  return true;
}

bool OptionTypeInfo::StructsAreEqual(
    const ConfigOptions& config_options, const std::string& struct_name,
    const std::unordered_map<std::string, OptionTypeInfo>* struct_map,
    const std::string& opt_name, const void* this_addr, const void* that_addr,
    std::string* mismatch) {
  assert(struct_map);
  bool matches = true;
  std::string result;
  if (EndsWith(opt_name, struct_name)) {
    // This option represents the entire struct
    matches = TypesAreEqual(config_options, *struct_map, this_addr, that_addr,
                            &result);
    if (!matches) {
      *mismatch = struct_name + "." + result;
      return false;
    }
  } else if (StartsWith(opt_name, struct_name + ".")) {
    // This option represents a nested field in the struct (e.g, struct.field)
    std::string elem_name;
    const auto opt_info =
        Find(opt_name.substr(struct_name.size() + 1), *struct_map, &elem_name);
    assert(opt_info);
    if (opt_info == nullptr) {
      *mismatch = opt_name;
      matches = false;
    } else if (!opt_info->AreEqual(config_options, elem_name, this_addr,
                                   that_addr, &result)) {
      matches = false;
      *mismatch = struct_name + "." + result;
    }
  } else {
    // This option represents a field in the struct (e.g. field)
    std::string elem_name;
    const auto opt_info = Find(opt_name, *struct_map, &elem_name);
    assert(opt_info);
    if (opt_info == nullptr) {
      *mismatch = struct_name + "." + opt_name;
      matches = false;
    } else if (!opt_info->AreEqual(config_options, elem_name, this_addr,
                                   that_addr, &result)) {
      matches = false;
      *mismatch = struct_name + "." + result;
    }
  }
  return matches;
}

bool MatchesOptionsTypeFromMap(
    const ConfigOptions& config_options,
    const std::unordered_map<std::string, OptionTypeInfo>& type_map,
    const void* const this_ptr, const void* const that_ptr,
    std::string* mismatch) {
  for (auto& pair : type_map) {
    // We skip checking deprecated variables as they might
    // contain random values since they might not be initialized
    if (config_options.IsCheckEnabled(pair.second.GetSanityLevel())) {
      if (!pair.second.AreEqual(config_options, pair.first, this_ptr, that_ptr,
                                mismatch) &&
          !pair.second.AreEqualByName(config_options, pair.first, this_ptr,
                                      that_ptr)) {
        return false;
      }
    }
  }
  return true;
}

bool OptionTypeInfo::AreEqualByName(const ConfigOptions& config_options,
                                    const std::string& opt_name,
                                    const void* const this_ptr,
                                    const void* const that_ptr) const {
  if (IsByName()) {
    std::string that_value;
    if (Serialize(config_options, opt_name, that_ptr, &that_value).ok()) {
      return AreEqualByName(config_options, opt_name, this_ptr, that_value);
    }
  }
  return false;
}

bool OptionTypeInfo::AreEqualByName(const ConfigOptions& config_options,
                                    const std::string& opt_name,
                                    const void* const opt_ptr,
                                    const std::string& that_value) const {
  std::string this_value;
  if (!IsByName()) {
    return false;
  } else if (!Serialize(config_options, opt_name, opt_ptr, &this_value).ok()) {
    return false;
  } else if (IsEnabled(OptionVerificationType::kByNameAllowFromNull)) {
    if (that_value == kNullptrString) {
      return true;
    }
  } else if (IsEnabled(OptionVerificationType::kByNameAllowNull)) {
    if (that_value == kNullptrString) {
      return true;
    }
  }
  return (this_value == that_value);
}

Status OptionTypeInfo::Prepare(const ConfigOptions& config_options,
                               const std::string& name, void* opt_ptr) const {
  if (ShouldPrepare()) {
    if (prepare_func_ != nullptr) {
      void* opt_addr = GetOffset(opt_ptr);
      return prepare_func_(config_options, name, opt_addr);
    } else if (IsConfigurable()) {
      Configurable* config = AsRawPointer<Configurable>(opt_ptr);
      if (config != nullptr) {
        return config->PrepareOptions(config_options);
      } else if (!CanBeNull()) {
        return Status::NotFound("Missing configurable object", name);
      }
    }
  }
  return Status::OK();
}

Status OptionTypeInfo::Validate(const DBOptions& db_opts,
                                const ColumnFamilyOptions& cf_opts,
                                const std::string& name,
                                const void* opt_ptr) const {
  if (ShouldValidate()) {
    if (validate_func_ != nullptr) {
      const void* opt_addr = GetOffset(opt_ptr);
      return validate_func_(db_opts, cf_opts, name, opt_addr);
    } else if (IsConfigurable()) {
      const Configurable* config = AsRawPointer<Configurable>(opt_ptr);
      if (config != nullptr) {
        return config->ValidateOptions(db_opts, cf_opts);
      } else if (!CanBeNull()) {
        return Status::NotFound("Missing configurable object", name);
      }
    }
  }
  return Status::OK();
}

const OptionTypeInfo* OptionTypeInfo::Find(
    const std::string& opt_name,
    const std::unordered_map<std::string, OptionTypeInfo>& opt_map,
    std::string* elem_name) {
  const auto iter = opt_map.find(opt_name);  // Look up the value in the map
  if (iter != opt_map.end()) {               // Found the option in the map
    *elem_name = opt_name;                   // Return the name
    return &(iter->second);  // Return the contents of the iterator
  } else {
    auto idx = opt_name.find(".");              // Look for a separator
    if (idx > 0 && idx != std::string::npos) {  // We found a separator
      auto siter =
          opt_map.find(opt_name.substr(0, idx));  // Look for the short name
      if (siter != opt_map.end()) {               // We found the short name
        if (siter->second.IsStruct() ||           // If the object is a struct
            siter->second.IsConfigurable()) {     // or a Configurable
          *elem_name = opt_name.substr(idx + 1);  // Return the rest
          return &(siter->second);  // Return the contents of the iterator
        }
      }
    }
  }
  return nullptr;
}

}  // namespace ROCKSDB_NAMESPACE