Make options.bottommost_compression, compression_opts and bottommost_compression_opts dynamically changeable. (#6615)

Summary:
These three options should be made dynamically changeable. Simply add them to MutableCFOptions and made the change.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6615

Test Plan: Add a unit test to make sure that SetOptions() can change the options.

Reviewed By: riversand963

Differential Revision: D20755951

fbshipit-source-id: 8165f4fd7a7a665cc7fb049698935022a5d2e7ff
main
sdong 5 years ago committed by Facebook GitHub Bot
parent fd3ddaf90d
commit 80979f81c7
  1. 1
      HISTORY.md
  2. 10
      db/compaction/compaction_job_test.cc
  3. 22
      db/compaction/compaction_picker.cc
  4. 8
      db/compaction/compaction_picker.h
  5. 12
      db/compaction/compaction_picker_fifo.cc
  6. 2
      db/compaction/compaction_picker_level.cc
  7. 6
      db/compaction/compaction_picker_universal.cc
  8. 2
      db/db_impl/db_impl_open.cc
  9. 56
      db/db_options_test.cc
  10. 2
      db/flush_job.cc
  11. 3
      options/cf_options.cc
  12. 14
      options/cf_options.h
  13. 187
      options/options_helper.cc
  14. 15
      table/sst_file_writer.cc

@ -24,6 +24,7 @@
* Basic support for user timestamp in iterator. Seek/SeekToFirst/Next and lower/upper bounds are supported. Reverse iteration is not supported. Merge is not considered. * Basic support for user timestamp in iterator. Seek/SeekToFirst/Next and lower/upper bounds are supported. Reverse iteration is not supported. Merge is not considered.
* When file lock failure when the lock is held by the current process, return acquiring time and thread ID in the error message. * When file lock failure when the lock is held by the current process, return acquiring time and thread ID in the error message.
* Added a new option, best_efforts_recovery (default: false), to allow database to open in a db dir with missing table files. During best efforts recovery, missing table files are ignored, and database recovers to the most recent state without missing table file. Cross-column-family consistency is not guaranteed even if WAL is enabled. * Added a new option, best_efforts_recovery (default: false), to allow database to open in a db dir with missing table files. During best efforts recovery, missing table files are ignored, and database recovers to the most recent state without missing table file. Cross-column-family consistency is not guaranteed even if WAL is enabled.
* options.bottommost_compression, options.compression_opts and options.bottommost_compression_opts are now dynamically changeable.
## 6.8.0 (02/24/2020) ## 6.8.0 (02/24/2020)
### Java API Changes ### Java API Changes

@ -314,11 +314,11 @@ class CompactionJobTest : public testing::Test {
num_input_files += level_files.size(); num_input_files += level_files.size();
} }
Compaction compaction(cfd->current()->storage_info(), *cfd->ioptions(), Compaction compaction(
*cfd->GetLatestMutableCFOptions(), cfd->current()->storage_info(), *cfd->ioptions(),
compaction_input_files, output_level, 1024 * 1024, *cfd->GetLatestMutableCFOptions(), compaction_input_files, output_level,
10 * 1024 * 1024, 0, kNoCompression, 1024 * 1024, 10 * 1024 * 1024, 0, kNoCompression,
cfd->ioptions()->compression_opts, 0, {}, true); cfd->GetLatestMutableCFOptions()->compression_opts, 0, {}, true);
compaction.SetInputVersion(cfd->current()); compaction.SetInputVersion(cfd->current());
LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, db_options_.info_log.get()); LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, db_options_.info_log.get());

@ -110,9 +110,9 @@ CompressionType GetCompressionType(const ImmutableCFOptions& ioptions,
// If bottommost_compression is set and we are compacting to the // If bottommost_compression is set and we are compacting to the
// bottommost level then we should use it. // bottommost level then we should use it.
if (ioptions.bottommost_compression != kDisableCompressionOption && if (mutable_cf_options.bottommost_compression != kDisableCompressionOption &&
level >= (vstorage->num_non_empty_levels() - 1)) { level >= (vstorage->num_non_empty_levels() - 1)) {
return ioptions.bottommost_compression; return mutable_cf_options.bottommost_compression;
} }
// If the user has specified a different compression level for each level, // If the user has specified a different compression level for each level,
// then pick the compression for that level. // then pick the compression for that level.
@ -132,22 +132,22 @@ CompressionType GetCompressionType(const ImmutableCFOptions& ioptions,
} }
} }
CompressionOptions GetCompressionOptions(const ImmutableCFOptions& ioptions, CompressionOptions GetCompressionOptions(const MutableCFOptions& cf_options,
const VersionStorageInfo* vstorage, const VersionStorageInfo* vstorage,
int level, int level,
const bool enable_compression) { const bool enable_compression) {
if (!enable_compression) { if (!enable_compression) {
return ioptions.compression_opts; return cf_options.compression_opts;
} }
// If bottommost_compression is set and we are compacting to the // If bottommost_compression is set and we are compacting to the
// bottommost level then we should use the specified compression options // bottommost level then we should use the specified compression options
// for the bottmomost_compression. // for the bottmomost_compression.
if (ioptions.bottommost_compression != kDisableCompressionOption && if (cf_options.bottommost_compression != kDisableCompressionOption &&
level >= (vstorage->num_non_empty_levels() - 1) && level >= (vstorage->num_non_empty_levels() - 1) &&
ioptions.bottommost_compression_opts.enabled) { cf_options.bottommost_compression_opts.enabled) {
return ioptions.bottommost_compression_opts; return cf_options.bottommost_compression_opts;
} }
return ioptions.compression_opts; return cf_options.compression_opts;
} }
CompactionPicker::CompactionPicker(const ImmutableCFOptions& ioptions, CompactionPicker::CompactionPicker(const ImmutableCFOptions& ioptions,
@ -359,7 +359,7 @@ Compaction* CompactionPicker::CompactFiles(
vstorage, ioptions_, mutable_cf_options, input_files, output_level, vstorage, ioptions_, mutable_cf_options, input_files, output_level,
compact_options.output_file_size_limit, compact_options.output_file_size_limit,
mutable_cf_options.max_compaction_bytes, output_path_id, compression_type, mutable_cf_options.max_compaction_bytes, output_path_id, compression_type,
GetCompressionOptions(ioptions_, vstorage, output_level), GetCompressionOptions(mutable_cf_options, vstorage, output_level),
compact_options.max_subcompactions, compact_options.max_subcompactions,
/* grandparents */ {}, true); /* grandparents */ {}, true);
RegisterCompaction(c); RegisterCompaction(c);
@ -634,7 +634,7 @@ Compaction* CompactionPicker::CompactRange(
compact_range_options.target_path_id, compact_range_options.target_path_id,
GetCompressionType(ioptions_, vstorage, mutable_cf_options, GetCompressionType(ioptions_, vstorage, mutable_cf_options,
output_level, 1), output_level, 1),
GetCompressionOptions(ioptions_, vstorage, output_level), GetCompressionOptions(mutable_cf_options, vstorage, output_level),
compact_range_options.max_subcompactions, /* grandparents */ {}, compact_range_options.max_subcompactions, /* grandparents */ {},
/* is manual */ true); /* is manual */ true);
RegisterCompaction(c); RegisterCompaction(c);
@ -787,7 +787,7 @@ Compaction* CompactionPicker::CompactRange(
compact_range_options.target_path_id, compact_range_options.target_path_id,
GetCompressionType(ioptions_, vstorage, mutable_cf_options, output_level, GetCompressionType(ioptions_, vstorage, mutable_cf_options, output_level,
vstorage->base_level()), vstorage->base_level()),
GetCompressionOptions(ioptions_, vstorage, output_level), GetCompressionOptions(mutable_cf_options, vstorage, output_level),
compact_range_options.max_subcompactions, std::move(grandparents), compact_range_options.max_subcompactions, std::move(grandparents),
/* is manual compaction */ true); /* is manual compaction */ true);

@ -305,9 +305,9 @@ CompressionType GetCompressionType(const ImmutableCFOptions& ioptions,
int level, int base_level, int level, int base_level,
const bool enable_compression = true); const bool enable_compression = true);
CompressionOptions GetCompressionOptions(const ImmutableCFOptions& ioptions, CompressionOptions GetCompressionOptions(
const VersionStorageInfo* vstorage, const MutableCFOptions& mutable_cf_options,
int level, const VersionStorageInfo* vstorage, int level,
const bool enable_compression = true); const bool enable_compression = true);
} // namespace ROCKSDB_NAMESPACE } // namespace ROCKSDB_NAMESPACE

@ -107,8 +107,9 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
Compaction* c = new Compaction( Compaction* c = new Compaction(
vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0, vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0,
kNoCompression, ioptions_.compression_opts, /* max_subcompactions */ 0, kNoCompression, mutable_cf_options.compression_opts,
{}, /* is manual */ false, vstorage->CompactionScore(0), /* max_subcompactions */ 0, {}, /* is manual */ false,
vstorage->CompactionScore(0),
/* is deletion compaction */ true, CompactionReason::kFIFOTtl); /* is deletion compaction */ true, CompactionReason::kFIFOTtl);
return c; return c;
} }
@ -148,7 +149,7 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
16 * 1024 * 1024 /* output file size limit */, 16 * 1024 * 1024 /* output file size limit */,
0 /* max compaction bytes, not applicable */, 0 /* max compaction bytes, not applicable */,
0 /* output path ID */, mutable_cf_options.compression, 0 /* output path ID */, mutable_cf_options.compression,
ioptions_.compression_opts, 0 /* max_subcompactions */, {}, mutable_cf_options.compression_opts, 0 /* max_subcompactions */, {},
/* is manual */ false, vstorage->CompactionScore(0), /* is manual */ false, vstorage->CompactionScore(0),
/* is deletion compaction */ false, /* is deletion compaction */ false,
CompactionReason::kFIFOReduceNumFiles); CompactionReason::kFIFOReduceNumFiles);
@ -196,8 +197,9 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
Compaction* c = new Compaction( Compaction* c = new Compaction(
vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0, vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0,
kNoCompression, ioptions_.compression_opts, /* max_subcompactions */ 0, kNoCompression, mutable_cf_options.compression_opts,
{}, /* is manual */ false, vstorage->CompactionScore(0), /* max_subcompactions */ 0, {}, /* is manual */ false,
vstorage->CompactionScore(0),
/* is deletion compaction */ true, CompactionReason::kFIFOMaxSize); /* is deletion compaction */ true, CompactionReason::kFIFOMaxSize);
return c; return c;
} }

@ -384,7 +384,7 @@ Compaction* LevelCompactionBuilder::GetCompaction() {
GetPathId(ioptions_, mutable_cf_options_, output_level_), GetPathId(ioptions_, mutable_cf_options_, output_level_),
GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, GetCompressionType(ioptions_, vstorage_, mutable_cf_options_,
output_level_, vstorage_->base_level()), output_level_, vstorage_->base_level()),
GetCompressionOptions(ioptions_, vstorage_, output_level_), GetCompressionOptions(mutable_cf_options_, vstorage_, output_level_),
/* max_subcompactions */ 0, std::move(grandparents_), is_manual_, /* max_subcompactions */ 0, std::move(grandparents_), is_manual_,
start_level_score_, false /* deletion_compaction */, compaction_reason_); start_level_score_, false /* deletion_compaction */, compaction_reason_);

@ -751,7 +751,7 @@ Compaction* UniversalCompactionBuilder::PickCompactionToReduceSortedRuns(
LLONG_MAX, path_id, LLONG_MAX, path_id,
GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, start_level, GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, start_level,
1, enable_compression), 1, enable_compression),
GetCompressionOptions(ioptions_, vstorage_, start_level, GetCompressionOptions(mutable_cf_options_, vstorage_, start_level,
enable_compression), enable_compression),
/* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false, /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false,
score_, false /* deletion_compaction */, compaction_reason); score_, false /* deletion_compaction */, compaction_reason);
@ -959,7 +959,7 @@ Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
/* max_grandparent_overlap_bytes */ LLONG_MAX, path_id, /* max_grandparent_overlap_bytes */ LLONG_MAX, path_id,
GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, GetCompressionType(ioptions_, vstorage_, mutable_cf_options_,
output_level, 1), output_level, 1),
GetCompressionOptions(ioptions_, vstorage_, output_level), GetCompressionOptions(mutable_cf_options_, vstorage_, output_level),
/* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ true, /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ true,
score_, false /* deletion_compaction */, score_, false /* deletion_compaction */,
CompactionReason::kFilesMarkedForCompaction); CompactionReason::kFilesMarkedForCompaction);
@ -1029,7 +1029,7 @@ Compaction* UniversalCompactionBuilder::PickCompactionToOldest(
LLONG_MAX, path_id, LLONG_MAX, path_id,
GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, start_level, GetCompressionType(ioptions_, vstorage_, mutable_cf_options_, start_level,
1, true /* enable_compression */), 1, true /* enable_compression */),
GetCompressionOptions(ioptions_, vstorage_, start_level, GetCompressionOptions(mutable_cf_options_, vstorage_, start_level,
true /* enable_compression */), true /* enable_compression */),
/* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false, /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false,
score_, false /* deletion_compaction */, compaction_reason); score_, false /* deletion_compaction */, compaction_reason);

@ -1248,7 +1248,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
snapshot_seqs, earliest_write_conflict_snapshot, snapshot_checker, snapshot_seqs, earliest_write_conflict_snapshot, snapshot_checker,
GetCompressionFlush(*cfd->ioptions(), mutable_cf_options), GetCompressionFlush(*cfd->ioptions(), mutable_cf_options),
mutable_cf_options.sample_for_compression, mutable_cf_options.sample_for_compression,
cfd->ioptions()->compression_opts, paranoid_file_checks, mutable_cf_options.compression_opts, paranoid_file_checks,
cfd->internal_stats(), TableFileCreationReason::kRecovery, &io_s, cfd->internal_stats(), TableFileCreationReason::kRecovery, &io_s,
&event_logger_, job_id, Env::IO_HIGH, nullptr /* table_properties */, &event_logger_, job_id, Env::IO_HIGH, nullptr /* table_properties */,
-1 /* level */, current_time, write_hint); -1 /* level */, current_time, write_hint);

@ -861,6 +861,62 @@ TEST_F(DBOptionsTest, FIFOTtlBackwardCompatible) {
#endif // ROCKSDB_LITE #endif // ROCKSDB_LITE
TEST_F(DBOptionsTest, ChangeCompression) {
if (!Snappy_Supported() || !LZ4_Supported()) {
return;
}
Options options;
options.write_buffer_size = 10 << 10; // 10KB
options.level0_file_num_compaction_trigger = 2;
options.create_if_missing = true;
options.compression = CompressionType::kLZ4Compression;
options.bottommost_compression = CompressionType::kNoCompression;
options.bottommost_compression_opts.level = 2;
ASSERT_OK(TryReopen(options));
CompressionType compression_used = CompressionType::kLZ4Compression;
CompressionOptions compression_opt_used;
bool compacted = false;
SyncPoint::GetInstance()->SetCallBack(
"LevelCompactionPicker::PickCompaction:Return", [&](void* arg) {
Compaction* c = reinterpret_cast<Compaction*>(arg);
compression_used = c->output_compression();
compression_opt_used = c->output_compression_opts();
compacted = true;
});
SyncPoint::GetInstance()->EnableProcessing();
ASSERT_OK(Put("foo", "foofoofoo"));
ASSERT_OK(Put("bar", "foofoofoo"));
ASSERT_OK(Flush());
ASSERT_OK(Put("foo", "foofoofoo"));
ASSERT_OK(Put("bar", "foofoofoo"));
ASSERT_OK(Flush());
dbfull()->TEST_WaitForCompact();
ASSERT_TRUE(compacted);
ASSERT_EQ(CompressionType::kNoCompression, compression_used);
ASSERT_EQ(options.compression_opts.level, compression_opt_used.level);
compression_used = CompressionType::kLZ4Compression;
compacted = false;
ASSERT_OK(dbfull()->SetOptions(
{{"bottommost_compression", "kSnappyCompression"},
{"bottommost_compression_opts", "0:6:0:0:0:true"}}));
ASSERT_OK(Put("foo", "foofoofoo"));
ASSERT_OK(Put("bar", "foofoofoo"));
ASSERT_OK(Flush());
ASSERT_OK(Put("foo", "foofoofoo"));
ASSERT_OK(Put("bar", "foofoofoo"));
ASSERT_OK(Flush());
dbfull()->TEST_WaitForCompact();
ASSERT_TRUE(compacted);
ASSERT_EQ(CompressionType::kSnappyCompression, compression_used);
ASSERT_EQ(6, compression_opt_used.level);
SyncPoint::GetInstance()->DisableProcessing();
}
} // namespace ROCKSDB_NAMESPACE } // namespace ROCKSDB_NAMESPACE
int main(int argc, char** argv) { int main(int argc, char** argv) {

@ -389,7 +389,7 @@ Status FlushJob::WriteLevel0Table() {
cfd_->GetName(), existing_snapshots_, cfd_->GetName(), existing_snapshots_,
earliest_write_conflict_snapshot_, snapshot_checker_, earliest_write_conflict_snapshot_, snapshot_checker_,
output_compression_, mutable_cf_options_.sample_for_compression, output_compression_, mutable_cf_options_.sample_for_compression,
cfd_->ioptions()->compression_opts, mutable_cf_options_.compression_opts,
mutable_cf_options_.paranoid_file_checks, cfd_->internal_stats(), mutable_cf_options_.paranoid_file_checks, cfd_->internal_stats(),
TableFileCreationReason::kFlush, &io_s, event_logger_, TableFileCreationReason::kFlush, &io_s, event_logger_,
job_context_->job_id, Env::IO_HIGH, &table_properties_, 0 /* level */, job_context_->job_id, Env::IO_HIGH, &table_properties_, 0 /* level */,

@ -57,9 +57,6 @@ ImmutableCFOptions::ImmutableCFOptions(const ImmutableDBOptions& db_options,
cf_options.purge_redundant_kvs_while_flush), cf_options.purge_redundant_kvs_while_flush),
use_fsync(db_options.use_fsync), use_fsync(db_options.use_fsync),
compression_per_level(cf_options.compression_per_level), compression_per_level(cf_options.compression_per_level),
bottommost_compression(cf_options.bottommost_compression),
bottommost_compression_opts(cf_options.bottommost_compression_opts),
compression_opts(cf_options.compression_opts),
level_compaction_dynamic_level_bytes( level_compaction_dynamic_level_bytes(
cf_options.level_compaction_dynamic_level_bytes), cf_options.level_compaction_dynamic_level_bytes),
access_hint_on_compaction_start( access_hint_on_compaction_start(

@ -90,12 +90,6 @@ struct ImmutableCFOptions {
std::vector<CompressionType> compression_per_level; std::vector<CompressionType> compression_per_level;
CompressionType bottommost_compression;
CompressionOptions bottommost_compression_opts;
CompressionOptions compression_opts;
bool level_compaction_dynamic_level_bytes; bool level_compaction_dynamic_level_bytes;
Options::AccessHint access_hint_on_compaction_start; Options::AccessHint access_hint_on_compaction_start;
@ -166,6 +160,9 @@ struct MutableCFOptions {
paranoid_file_checks(options.paranoid_file_checks), paranoid_file_checks(options.paranoid_file_checks),
report_bg_io_stats(options.report_bg_io_stats), report_bg_io_stats(options.report_bg_io_stats),
compression(options.compression), compression(options.compression),
bottommost_compression(options.bottommost_compression),
compression_opts(options.compression_opts),
bottommost_compression_opts(options.bottommost_compression_opts),
sample_for_compression(options.sample_for_compression) { sample_for_compression(options.sample_for_compression) {
RefreshDerivedOptions(options.num_levels, options.compaction_style); RefreshDerivedOptions(options.num_levels, options.compaction_style);
} }
@ -198,6 +195,7 @@ struct MutableCFOptions {
paranoid_file_checks(false), paranoid_file_checks(false),
report_bg_io_stats(false), report_bg_io_stats(false),
compression(Snappy_Supported() ? kSnappyCompression : kNoCompression), compression(Snappy_Supported() ? kSnappyCompression : kNoCompression),
bottommost_compression(kDisableCompressionOption),
sample_for_compression(0) {} sample_for_compression(0) {}
explicit MutableCFOptions(const Options& options); explicit MutableCFOptions(const Options& options);
@ -253,6 +251,10 @@ struct MutableCFOptions {
bool paranoid_file_checks; bool paranoid_file_checks;
bool report_bg_io_stats; bool report_bg_io_stats;
CompressionType compression; CompressionType compression;
CompressionType bottommost_compression;
CompressionOptions compression_opts;
CompressionOptions bottommost_compression_opts;
uint64_t sample_for_compression; uint64_t sample_for_compression;
// Derived options // Derived options

@ -259,6 +259,8 @@ std::unordered_map<std::string, CompressionType>
const std::string kNameComparator = "comparator"; const std::string kNameComparator = "comparator";
const std::string kNameEnv = "env"; const std::string kNameEnv = "env";
const std::string kNameMergeOperator = "merge_operator"; const std::string kNameMergeOperator = "merge_operator";
const std::string kOptNameBMCompOpts = "bottommost_compression_opts";
const std::string kOptNameCompOpts = "compression_opts";
template <typename T> template <typename T>
Status GetStringFromStruct( Status GetStringFromStruct(
@ -786,6 +788,66 @@ bool SerializeSingleOptionHelper(const char* opt_address,
return true; return true;
} }
Status ParseCompressionOptions(const std::string& value,
const std::string& name,
CompressionOptions& compression_opts) {
size_t start = 0;
size_t end = value.find(':');
if (end == std::string::npos) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
compression_opts.window_bits = ParseInt(value.substr(start, end - start));
start = end + 1;
end = value.find(':', start);
if (end == std::string::npos) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
compression_opts.level = ParseInt(value.substr(start, end - start));
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
end = value.find(':', start);
compression_opts.strategy =
ParseInt(value.substr(start, value.size() - start));
// max_dict_bytes is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.max_dict_bytes =
ParseInt(value.substr(start, value.size() - start));
end = value.find(':', start);
}
// zstd_max_train_bytes is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.zstd_max_train_bytes =
ParseInt(value.substr(start, value.size() - start));
end = value.find(':', start);
}
// enabled is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.enabled =
ParseBoolean("", value.substr(start, value.size() - start));
}
return Status::OK();
}
Status GetMutableOptionsFromStrings( Status GetMutableOptionsFromStrings(
const MutableCFOptions& base_options, const MutableCFOptions& base_options,
const std::unordered_map<std::string, std::string>& options_map, const std::unordered_map<std::string, std::string>& options_map,
@ -793,30 +855,50 @@ Status GetMutableOptionsFromStrings(
assert(new_options); assert(new_options);
*new_options = base_options; *new_options = base_options;
for (const auto& o : options_map) { for (const auto& o : options_map) {
auto& option_name = o.first;
auto& option_value = o.second;
try { try {
auto iter = cf_options_type_info.find(o.first); if (option_name == kOptNameBMCompOpts) {
if (iter == cf_options_type_info.end()) { Status s =
return Status::InvalidArgument("Unrecognized option: " + o.first); ParseCompressionOptions(option_value, option_name,
} new_options->bottommost_compression_opts);
const auto& opt_info = iter->second; if (!s.ok()) {
if (!opt_info.is_mutable) { return s;
return Status::InvalidArgument("Option not changeable: " + o.first); }
} } else if (option_name == kOptNameCompOpts) {
if (opt_info.verification == OptionVerificationType::kDeprecated) { Status s = ParseCompressionOptions(option_value, option_name,
// log warning when user tries to set a deprecated option but don't fail new_options->compression_opts);
// the call for compatibility. if (!s.ok()) {
ROCKS_LOG_WARN(info_log, "%s is a deprecated option and cannot be set", return s;
o.first.c_str()); }
continue; } else {
} auto iter = cf_options_type_info.find(option_name);
bool is_ok = ParseOptionHelper( if (iter == cf_options_type_info.end()) {
reinterpret_cast<char*>(new_options) + opt_info.mutable_offset, return Status::InvalidArgument("Unrecognized option: " + option_name);
opt_info.type, o.second); }
if (!is_ok) { const auto& opt_info = iter->second;
return Status::InvalidArgument("Error parsing " + o.first); if (!opt_info.is_mutable) {
return Status::InvalidArgument("Option not changeable: " +
option_name);
}
if (opt_info.verification == OptionVerificationType::kDeprecated) {
// log warning when user tries to set a deprecated option but don't
// fail the call for compatibility.
ROCKS_LOG_WARN(info_log,
"%s is a deprecated option and cannot be set",
option_name.c_str());
continue;
}
bool is_ok = ParseOptionHelper(
reinterpret_cast<char*>(new_options) + opt_info.mutable_offset,
opt_info.type, option_value);
if (!is_ok) {
return Status::InvalidArgument("Error parsing " + option_name);
}
} }
} catch (std::exception& e) { } catch (std::exception& e) {
return Status::InvalidArgument("Error parsing " + o.first + ":" + return Status::InvalidArgument("Error parsing " + option_name + ":" +
std::string(e.what())); std::string(e.what()));
} }
} }
@ -929,65 +1011,6 @@ Status StringToMap(const std::string& opts_str,
return Status::OK(); return Status::OK();
} }
Status ParseCompressionOptions(const std::string& value, const std::string& name,
CompressionOptions& compression_opts) {
size_t start = 0;
size_t end = value.find(':');
if (end == std::string::npos) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
compression_opts.window_bits = ParseInt(value.substr(start, end - start));
start = end + 1;
end = value.find(':', start);
if (end == std::string::npos) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
compression_opts.level = ParseInt(value.substr(start, end - start));
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument("unable to parse the specified CF option " +
name);
}
end = value.find(':', start);
compression_opts.strategy =
ParseInt(value.substr(start, value.size() - start));
// max_dict_bytes is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.max_dict_bytes =
ParseInt(value.substr(start, value.size() - start));
end = value.find(':', start);
}
// zstd_max_train_bytes is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.zstd_max_train_bytes =
ParseInt(value.substr(start, value.size() - start));
end = value.find(':', start);
}
// enabled is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
compression_opts.enabled =
ParseBoolean("", value.substr(start, value.size() - start));
}
return Status::OK();
}
Status ParseColumnFamilyOption(const std::string& name, Status ParseColumnFamilyOption(const std::string& name,
const std::string& org_value, const std::string& org_value,
ColumnFamilyOptions* new_options, ColumnFamilyOptions* new_options,
@ -1986,8 +2009,8 @@ std::unordered_map<std::string, OptionTypeInfo>
false, 0}}, false, 0}},
{"bottommost_compression", {"bottommost_compression",
{offset_of(&ColumnFamilyOptions::bottommost_compression), {offset_of(&ColumnFamilyOptions::bottommost_compression),
OptionType::kCompressionType, OptionVerificationType::kNormal, false, OptionType::kCompressionType, OptionVerificationType::kNormal, true,
0}}, offsetof(struct MutableCFOptions, bottommost_compression)}},
{kNameComparator, {kNameComparator,
{offset_of(&ColumnFamilyOptions::comparator), OptionType::kComparator, {offset_of(&ColumnFamilyOptions::comparator), OptionType::kComparator,
OptionVerificationType::kByName, false, 0}}, OptionVerificationType::kByName, false, 0}},

@ -190,20 +190,21 @@ Status SstFileWriter::Open(const std::string& file_path) {
CompressionType compression_type; CompressionType compression_type;
CompressionOptions compression_opts; CompressionOptions compression_opts;
if (r->ioptions.bottommost_compression != kDisableCompressionOption) { if (r->mutable_cf_options.bottommost_compression !=
compression_type = r->ioptions.bottommost_compression; kDisableCompressionOption) {
if (r->ioptions.bottommost_compression_opts.enabled) { compression_type = r->mutable_cf_options.bottommost_compression;
compression_opts = r->ioptions.bottommost_compression_opts; if (r->mutable_cf_options.bottommost_compression_opts.enabled) {
compression_opts = r->mutable_cf_options.bottommost_compression_opts;
} else { } else {
compression_opts = r->ioptions.compression_opts; compression_opts = r->mutable_cf_options.compression_opts;
} }
} else if (!r->ioptions.compression_per_level.empty()) { } else if (!r->ioptions.compression_per_level.empty()) {
// Use the compression of the last level if we have per level compression // Use the compression of the last level if we have per level compression
compression_type = *(r->ioptions.compression_per_level.rbegin()); compression_type = *(r->ioptions.compression_per_level.rbegin());
compression_opts = r->ioptions.compression_opts; compression_opts = r->mutable_cf_options.compression_opts;
} else { } else {
compression_type = r->mutable_cf_options.compression; compression_type = r->mutable_cf_options.compression;
compression_opts = r->ioptions.compression_opts; compression_opts = r->mutable_cf_options.compression_opts;
} }
uint64_t sample_for_compression = uint64_t sample_for_compression =
r->mutable_cf_options.sample_for_compression; r->mutable_cf_options.sample_for_compression;

Loading…
Cancel
Save