Add Support for saving CompressionOptions to Options File (#6817)

Summary: This PR does a few things: - The "compression_opts" and "bottom_compression_opts" can now be read/written as name/value pairs of options (instead of only a colon-separated list; - These options can now be read/written to the Options file; - The parallel_threads value can now be set (either in the colon or name-value format). The compression options are now stored and treated as a OptionTypeInfo::Struct by the options system, meaning they can be read and written like the other structs. This change allows them to be read/written easily to the options file. Additionally, the colon-format was extended to allow support for setting parallel threads. Tests were added to test all of the option settings via the optional parameters in the colon format. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6817 Reviewed By: ajkr Differential Revision: D22396004 Pulled By: zhichao-cao fbshipit-source-id: 38bcf74b7e9cd5bc2a84540fac2e9ba4f765b2c8
4 years ago · ec711b2315
parent c5ddeceba0
commit ec711b2315
2 changed files with 266 additions and 25 deletions
--- a/options/cf_options.cc
+++ b/options/cf_options.cc
@ -94,9 +94,29 @@ static Status ParseCompressionOptions(const std::string& value,
        ParseInt(value.substr(start, value.size() - start));
    end = value.find(':', start);
  }
-  // parallel_threads is not serialized with this format.
-  // We plan to upgrade the format to a JSON-like format.
-  compression_opts.parallel_threads = CompressionOptions().parallel_threads;
+
+  // parallel_threads 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);
+    }
+    // Since parallel_threads comes before enabled but was added optionally
+    // later, we need to check if this is the final token (meaning it is the
+    // enabled bit), or if there is another token (meaning this one is
+    // parallel_threads)
+    end = value.find(':', start);
+    if (end != std::string::npos) {
+      compression_opts.parallel_threads =
+          ParseInt(value.substr(start, value.size() - start));
+    } else {
+      // parallel_threads is not serialized with this format, but enabled is
+      compression_opts.parallel_threads = CompressionOptions().parallel_threads;
+      compression_opts.enabled =
+          ParseBoolean("", value.substr(start, value.size() - start));
+    }
+  }

  // enabled is optional for backwards compatibility
  if (end != std::string::npos) {
@ -114,6 +134,41 @@ static Status ParseCompressionOptions(const std::string& value,
 const std::string kOptNameBMCompOpts = "bottommost_compression_opts";
 const std::string kOptNameCompOpts = "compression_opts";

+// OptionTypeInfo map for CompressionOptions
+static std::unordered_map<std::string, OptionTypeInfo>
+    compression_options_type_info = {
+        {"window_bits",
+         {offsetof(struct CompressionOptions, window_bits), OptionType::kInt,
+          OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, window_bits)}},
+        {"level",
+         {offsetof(struct CompressionOptions, level), OptionType::kInt,
+          OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, level)}},
+        {"strategy",
+         {offsetof(struct CompressionOptions, strategy), OptionType::kInt,
+          OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, strategy)}},
+        {"max_dict_bytes",
+         {offsetof(struct CompressionOptions, max_dict_bytes), OptionType::kInt,
+          OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, max_dict_bytes)}},
+        {"zstd_max_train_bytes",
+         {offsetof(struct CompressionOptions, zstd_max_train_bytes),
+          OptionType::kUInt32T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, zstd_max_train_bytes)}},
+        {"parallel_threads",
+         {offsetof(struct CompressionOptions, parallel_threads),
+          OptionType::kUInt32T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, parallel_threads)}},
+        {"enabled",
+         {offsetof(struct CompressionOptions, enabled), OptionType::kBoolean,
+          OptionVerificationType::kNormal, OptionTypeFlags::kMutable,
+          offsetof(struct CompressionOptions, enabled)}},
+};
+
 static std::unordered_map<std::string, OptionTypeInfo>
    fifo_compaction_options_type_info = {
        {"max_table_files_size",
@ -580,29 +635,49 @@ std::unordered_map<std::string, OptionTypeInfo>
        // This means that the properties could be read from the options file
        // but never written to the file or compared to each other.
        {kOptNameCompOpts,
-         {offset_of(&ColumnFamilyOptions::compression_opts),
-          OptionType::kUnknown, OptionVerificationType::kNormal,
-          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever |
-           OptionTypeFlags::kMutable),
-          offsetof(struct MutableCFOptions, compression_opts),
-          // Parses the value as a CompressionOptions
-          [](const ConfigOptions& /*opts*/, const std::string& name,
-             const std::string& value, char* addr) {
-            auto* compression = reinterpret_cast<CompressionOptions*>(addr);
-            return ParseCompressionOptions(value, name, *compression);
-          }}},
+         OptionTypeInfo::Struct(
+             kOptNameCompOpts, &compression_options_type_info,
+             offset_of(&ColumnFamilyOptions::compression_opts),
+             OptionVerificationType::kNormal,
+             (OptionTypeFlags::kMutable | OptionTypeFlags::kCompareNever),
+             offsetof(struct MutableCFOptions, compression_opts),
+             [](const ConfigOptions& opts, const std::string& name,
+                const std::string& value, char* addr) {
+               // This is to handle backward compatibility, where
+               // compression_options was a ":" separated list.
+               if (name == kOptNameCompOpts &&
+                   value.find("=") == std::string::npos) {
+                 auto* compression =
+                     reinterpret_cast<CompressionOptions*>(addr);
+                 return ParseCompressionOptions(value, name, *compression);
+               } else {
+                 return OptionTypeInfo::ParseStruct(
+                     opts, kOptNameCompOpts, &compression_options_type_info,
+                     name, value, addr);
+               }
+             })},
        {kOptNameBMCompOpts,
-         {offset_of(&ColumnFamilyOptions::bottommost_compression_opts),
-          OptionType::kUnknown, OptionVerificationType::kNormal,
-          (OptionTypeFlags::kDontSerialize | OptionTypeFlags::kCompareNever |
-           OptionTypeFlags::kMutable),
-          offsetof(struct MutableCFOptions, bottommost_compression_opts),
-          // Parses the value as a CompressionOptions
-          [](const ConfigOptions& /*opts*/, const std::string& name,
-             const std::string& value, char* addr) {
-            auto* compression = reinterpret_cast<CompressionOptions*>(addr);
-            return ParseCompressionOptions(value, name, *compression);
-          }}},
+         OptionTypeInfo::Struct(
+             kOptNameBMCompOpts, &compression_options_type_info,
+             offset_of(&ColumnFamilyOptions::bottommost_compression_opts),
+             OptionVerificationType::kNormal,
+             (OptionTypeFlags::kMutable | OptionTypeFlags::kCompareNever),
+             offsetof(struct MutableCFOptions, bottommost_compression_opts),
+             [](const ConfigOptions& opts, const std::string& name,
+                const std::string& value, char* addr) {
+               // This is to handle backward compatibility, where
+               // compression_options was a ":" separated list.
+               if (name == kOptNameBMCompOpts &&
+                   value.find("=") == std::string::npos) {
+                 auto* compression =
+                     reinterpret_cast<CompressionOptions*>(addr);
+                 return ParseCompressionOptions(value, name, *compression);
+               } else {
+                 return OptionTypeInfo::ParseStruct(
+                     opts, kOptNameBMCompOpts, &compression_options_type_info,
+                     name, value, addr);
+               }
+             })},
        // End special case properties
 };

--- a/options/options_test.cc
+++ b/options/options_test.cc
@ -553,6 +553,172 @@ TEST_F(OptionsTest, GetColumnFamilyOptionsFromStringTest) {
  ASSERT_EQ(std::string(new_cf_opt.memtable_factory->Name()), "SkipListFactory");
 }

+TEST_F(OptionsTest, CompressionOptionsFromString) {
+  ColumnFamilyOptions base_cf_opt;
+  ColumnFamilyOptions new_cf_opt;
+  ConfigOptions config_options;
+  std::string opts_str;
+  config_options.ignore_unknown_options = false;
+  CompressionOptions dflt;
+  // Test with some optional values removed....
+  ASSERT_OK(
+      GetColumnFamilyOptionsFromString(config_options, ColumnFamilyOptions(),
+                                       "compression_opts=3:4:5; "
+                                       "bottommost_compression_opts=4:5:6:7",
+                                       &base_cf_opt));
+  ASSERT_EQ(base_cf_opt.compression_opts.window_bits, 3);
+  ASSERT_EQ(base_cf_opt.compression_opts.level, 4);
+  ASSERT_EQ(base_cf_opt.compression_opts.strategy, 5);
+  ASSERT_EQ(base_cf_opt.compression_opts.max_dict_bytes, dflt.max_dict_bytes);
+  ASSERT_EQ(base_cf_opt.compression_opts.zstd_max_train_bytes,
+            dflt.zstd_max_train_bytes);
+  ASSERT_EQ(base_cf_opt.compression_opts.parallel_threads,
+            dflt.parallel_threads);
+  ASSERT_EQ(base_cf_opt.compression_opts.enabled, dflt.enabled);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.window_bits, 4);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.level, 5);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.strategy, 6);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.max_dict_bytes, 7u);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.zstd_max_train_bytes,
+            dflt.zstd_max_train_bytes);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.parallel_threads,
+            dflt.parallel_threads);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.enabled, dflt.enabled);
+
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(),
+      "compression_opts=4:5:6:7:8:9:true; "
+      "bottommost_compression_opts=5:6:7:8:9:false",
+      &base_cf_opt));
+  ASSERT_EQ(base_cf_opt.compression_opts.window_bits, 4);
+  ASSERT_EQ(base_cf_opt.compression_opts.level, 5);
+  ASSERT_EQ(base_cf_opt.compression_opts.strategy, 6);
+  ASSERT_EQ(base_cf_opt.compression_opts.max_dict_bytes, 7u);
+  ASSERT_EQ(base_cf_opt.compression_opts.zstd_max_train_bytes, 8u);
+  ASSERT_EQ(base_cf_opt.compression_opts.parallel_threads, 9u);
+  ASSERT_EQ(base_cf_opt.compression_opts.enabled, true);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.window_bits, 5);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.level, 6);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.strategy, 7);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.max_dict_bytes, 8u);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.zstd_max_train_bytes, 9u);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.parallel_threads,
+            dflt.parallel_threads);
+  ASSERT_EQ(base_cf_opt.bottommost_compression_opts.enabled, false);
+
+  ASSERT_OK(
+      GetStringFromColumnFamilyOptions(config_options, base_cf_opt, &opts_str));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(), opts_str, &new_cf_opt));
+  ASSERT_EQ(new_cf_opt.compression_opts.window_bits, 4);
+  ASSERT_EQ(new_cf_opt.compression_opts.level, 5);
+  ASSERT_EQ(new_cf_opt.compression_opts.strategy, 6);
+  ASSERT_EQ(new_cf_opt.compression_opts.max_dict_bytes, 7u);
+  ASSERT_EQ(new_cf_opt.compression_opts.zstd_max_train_bytes, 8u);
+  ASSERT_EQ(new_cf_opt.compression_opts.parallel_threads, 9u);
+  ASSERT_EQ(new_cf_opt.compression_opts.enabled, true);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.window_bits, 5);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.level, 6);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.strategy, 7);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.max_dict_bytes, 8u);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.zstd_max_train_bytes, 9u);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.parallel_threads,
+            dflt.parallel_threads);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.enabled, false);
+
+  // Test as struct values
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(),
+      "compression_opts={window_bits=5; level=6; strategy=7; max_dict_bytes=8;"
+      "zstd_max_train_bytes=9;parallel_threads=10;enabled=true}; "
+      "bottommost_compression_opts={window_bits=4; level=5; strategy=6;"
+      " max_dict_bytes=7;zstd_max_train_bytes=8;parallel_threads=9;"
+      "enabled=false}; ",
+      &new_cf_opt));
+  ASSERT_EQ(new_cf_opt.compression_opts.window_bits, 5);
+  ASSERT_EQ(new_cf_opt.compression_opts.level, 6);
+  ASSERT_EQ(new_cf_opt.compression_opts.strategy, 7);
+  ASSERT_EQ(new_cf_opt.compression_opts.max_dict_bytes, 8u);
+  ASSERT_EQ(new_cf_opt.compression_opts.zstd_max_train_bytes, 9u);
+  ASSERT_EQ(new_cf_opt.compression_opts.parallel_threads, 10u);
+  ASSERT_EQ(new_cf_opt.compression_opts.enabled, true);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.window_bits, 4);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.level, 5);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.strategy, 6);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.max_dict_bytes, 7u);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.zstd_max_train_bytes, 8u);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.parallel_threads, 9u);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.enabled, false);
+
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, base_cf_opt,
+      "compression_opts={window_bits=4; strategy=5;};"
+      "bottommost_compression_opts={level=6; strategy=7;}",
+      &new_cf_opt));
+  ASSERT_EQ(new_cf_opt.compression_opts.window_bits, 4);
+  ASSERT_EQ(new_cf_opt.compression_opts.strategy, 5);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.level, 6);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.strategy, 7);
+
+  ASSERT_EQ(new_cf_opt.compression_opts.level,
+            base_cf_opt.compression_opts.level);
+  ASSERT_EQ(new_cf_opt.compression_opts.max_dict_bytes,
+            base_cf_opt.compression_opts.max_dict_bytes);
+  ASSERT_EQ(new_cf_opt.compression_opts.zstd_max_train_bytes,
+            base_cf_opt.compression_opts.zstd_max_train_bytes);
+  ASSERT_EQ(new_cf_opt.compression_opts.parallel_threads,
+            base_cf_opt.compression_opts.parallel_threads);
+  ASSERT_EQ(new_cf_opt.compression_opts.enabled,
+            base_cf_opt.compression_opts.enabled);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.window_bits,
+            base_cf_opt.bottommost_compression_opts.window_bits);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.max_dict_bytes,
+            base_cf_opt.bottommost_compression_opts.max_dict_bytes);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.zstd_max_train_bytes,
+            base_cf_opt.bottommost_compression_opts.zstd_max_train_bytes);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.parallel_threads,
+            base_cf_opt.bottommost_compression_opts.parallel_threads);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.enabled,
+            base_cf_opt.bottommost_compression_opts.enabled);
+
+  // Test a few individual struct values
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      config_options, base_cf_opt,
+      "compression_opts.enabled=false; "
+      "bottommost_compression_opts.enabled=true; ",
+      &new_cf_opt));
+  ASSERT_EQ(new_cf_opt.compression_opts.enabled, false);
+  ASSERT_EQ(new_cf_opt.bottommost_compression_opts.enabled, true);
+
+  // Now test some illegal values
+  ConfigOptions ignore;
+  ignore.ignore_unknown_options = true;
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(),
+      "compression_opts=5:6:7:8:9:x:false", &base_cf_opt));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      ignore, ColumnFamilyOptions(), "compression_opts=5:6:7:8:9:x:false",
+      &base_cf_opt));
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(),
+      "compression_opts=1:2:3:4:5:6:true:8", &base_cf_opt));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(
+      ignore, ColumnFamilyOptions(), "compression_opts=1:2:3:4:5:6:true:8",
+      &base_cf_opt));
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(), "compression_opts={unknown=bad;}",
+      &base_cf_opt));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(ignore, ColumnFamilyOptions(),
+                                             "compression_opts={unknown=bad;}",
+                                             &base_cf_opt));
+  ASSERT_NOK(GetColumnFamilyOptionsFromString(
+      config_options, ColumnFamilyOptions(), "compression_opts.unknown=bad",
+      &base_cf_opt));
+  ASSERT_OK(GetColumnFamilyOptionsFromString(ignore, ColumnFamilyOptions(),
+                                             "compression_opts.unknown=bad",
+                                             &base_cf_opt));
+}
+
 TEST_F(OptionsTest, OldInterfaceTest) {
  ColumnFamilyOptions base_cf_opt;
  ColumnFamilyOptions new_cf_opt;