Initial add UDT in memtable only option (#11362)

Summary:
This option is immutable through the life time of the DB open. For now, updating its value between different DB open sessions is also a non compatible change. When I work on support for updating comparator, the type of updates accepted for this option will be supported then.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11362

Test Plan: `make check`

Reviewed By: ltamasi

Differential Revision: D44873870

Pulled By: jowlyzhang

fbshipit-source-id: aa02094754b58d99abf9af4c9a8108c1350254cb

include/rocksdb/advanced_options.h

@@ -1109,6 +1109,23 @@ struct AdvancedColumnFamilyOptions {
   // Supported values: 0, 1, 2, 4, 8.
   uint32_t memtable_protection_bytes_per_key = 0;
 
+  // UNDER CONSTRUCTION -- DO NOT USE
+  // When the user-defined timestamp feature is enabled, this flag controls
+  // whether the user-defined timestamps will be persisted.
+  //
+  // When it's false, the user-defined timestamps will be removed from the user
+  // keys when data is flushed from memtables to SST files. Other places that
+  // user keys can be persisted like WAL and blob files go through a similar
+  // process. Users should call `DB::IncreaseFullHistoryTsLow` to set a cutoff
+  // timestamp. RocksDB refrains from flushing a memtable with data still above
+  // the cutoff timestamp with best effort. When users try to read below the
+  // cutoff timestamp, an error will be returned.
+  //
+  // Default: true (user-defined timestamps are persisted)
+  // Not dynamically changeable, change it requires db restart and
+  // only compatible changes are allowed.
+  bool persist_user_defined_timestamps = true;
+
   // Create ColumnFamilyOptions with default values for all fields
   AdvancedColumnFamilyOptions();
   // Create ColumnFamilyOptions from Options

include/rocksdb/utilities/options_util.h

@@ -96,6 +96,7 @@ Status GetLatestOptionsFileName(const std::string& dbpath, Env* env,
 // * prefix_extractor
 // * table_factory
 // * merge_operator
+// * persist_user_defined_timestamps
 Status CheckOptionsCompatibility(
     const ConfigOptions& config_options, const std::string& dbpath,
     const DBOptions& db_options,

options/cf_options.cc

@@ -775,6 +775,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
             auto* cache = static_cast<std::shared_ptr<Cache>*>(addr);
             return Cache::CreateFromString(opts, value, cache);
           }}},
+        {"persist_user_defined_timestamps",
+         {offsetof(struct ImmutableCFOptions, persist_user_defined_timestamps),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kCompareLoose}},
 };
 
 const std::string OptionsHelper::kCFOptionsName = "ColumnFamilyOptions";
@@ -918,7 +922,9 @@ ImmutableCFOptions::ImmutableCFOptions(const ColumnFamilyOptions& cf_options)
       cf_paths(cf_options.cf_paths),
       compaction_thread_limiter(cf_options.compaction_thread_limiter),
       sst_partitioner_factory(cf_options.sst_partitioner_factory),
-      blob_cache(cf_options.blob_cache) {}
+      blob_cache(cf_options.blob_cache),
+      persist_user_defined_timestamps(
+          cf_options.persist_user_defined_timestamps) {}
 
 ImmutableOptions::ImmutableOptions() : ImmutableOptions(Options()) {}
 

options/cf_options.h

@@ -86,6 +86,8 @@ struct ImmutableCFOptions {
   std::shared_ptr<SstPartitionerFactory> sst_partitioner_factory;
 
   std::shared_ptr<Cache> blob_cache;
+
+  bool persist_user_defined_timestamps;
 };
 
 struct ImmutableOptions : public ImmutableDBOptions, public ImmutableCFOptions {

options/options.cc

@@ -109,7 +109,8 @@ AdvancedColumnFamilyOptions::AdvancedColumnFamilyOptions(const Options& options)
       blob_compaction_readahead_size(options.blob_compaction_readahead_size),
       blob_file_starting_level(options.blob_file_starting_level),
       blob_cache(options.blob_cache),
-      prepopulate_blob_cache(options.prepopulate_blob_cache) {
+      prepopulate_blob_cache(options.prepopulate_blob_cache),
+      persist_user_defined_timestamps(options.persist_user_defined_timestamps) {
   assert(memtable_factory.get() != nullptr);
   if (max_bytes_for_level_multiplier_additional.size() <
       static_cast<unsigned int>(num_levels)) {
@@ -137,6 +138,11 @@ void DBOptions::Dump(Logger* log) const {
 void ColumnFamilyOptions::Dump(Logger* log) const {
   ROCKS_LOG_HEADER(log, "              Options.comparator: %s",
                    comparator->Name());
+  if (comparator->timestamp_size() > 0) {
+    ROCKS_LOG_HEADER(
+        log, "              Options.persist_user_defined_timestamps: %s",
+        persist_user_defined_timestamps ? "true" : "false");
+  }
   ROCKS_LOG_HEADER(log, "          Options.merge_operator: %s",
                    merge_operator ? merge_operator->Name() : "None");
   ROCKS_LOG_HEADER(log, "       Options.compaction_filter: %s",

options/options_helper.cc

@@ -308,6 +308,8 @@ void UpdateColumnFamilyOptions(const ImmutableCFOptions& ioptions,
       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;
 
   // TODO(yhchiang): find some way to handle the following derived options
   // * max_file_size

options/options_settable_test.cc

@@ -546,7 +546,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "compaction_options_fifo={max_table_files_size=3;allow_"
       "compaction=false;age_for_warm=1;};"
       "blob_cache=1M;"
-      "memtable_protection_bytes_per_key=2;",
+      "memtable_protection_bytes_per_key=2;"
+      "persist_user_defined_timestamps=true;",
       new_options));
 
   ASSERT_NE(new_options->blob_cache.get(), nullptr);

options/options_test.cc

@@ -128,6 +128,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
       {"blob_file_starting_level", "1"},
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
+      {"persist_user_defined_timestamps", "true"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -269,6 +270,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.prepopulate_blob_cache, PrepopulateBlobCache::kDisable);
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
+  ASSERT_EQ(new_cf_opt.persist_user_defined_timestamps, true);
 
   cf_options_map["write_buffer_size"] = "hello";
   ASSERT_NOK(GetColumnFamilyOptionsFromMap(exact, base_cf_opt, cf_options_map,
@@ -2320,6 +2322,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
       {"blob_file_starting_level", "1"},
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
+      {"persist_user_defined_timestamps", "true"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -2459,6 +2462,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.prepopulate_blob_cache, PrepopulateBlobCache::kDisable);
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
+  ASSERT_EQ(new_cf_opt.persist_user_defined_timestamps, true);
 
   cf_options_map["write_buffer_size"] = "hello";
   ASSERT_NOK(GetColumnFamilyOptionsFromMap(cf_config_options, base_cf_opt,
@@ -4068,6 +4072,30 @@ TEST_P(OptionsSanityCheckTest, CFOptionsSanityCheck) {
       SanityCheckCFOptions(opts, config_options_.ignore_unsupported_options);
     }
   }
+
+  // persist_user_defined_timestamps
+  {
+    // Test change from true to false not allowed in loose and exact mode.
+    opts.persist_user_defined_timestamps = false;
+    ASSERT_NOK(SanityCheckCFOptions(
+        opts, ConfigOptions::kSanityLevelLooselyCompatible));
+    ASSERT_NOK(
+        SanityCheckCFOptions(opts, ConfigOptions::kSanityLevelExactMatch));
+
+    // persist the change
+    ASSERT_OK(PersistCFOptions(opts));
+    SanityCheckCFOptions(opts, config_options_.ignore_unsupported_options);
+
+    // Test change from false to true not allowed in loose and exact mode.
+    opts.persist_user_defined_timestamps = true;
+    ASSERT_NOK(SanityCheckCFOptions(
+        opts, ConfigOptions::kSanityLevelLooselyCompatible));
+    ASSERT_NOK(
+        SanityCheckCFOptions(opts, ConfigOptions::kSanityLevelExactMatch));
+
+    // persist the change
+    ASSERT_OK(PersistCFOptions(opts));
+  }
 }
 
 TEST_P(OptionsSanityCheckTest, DBOptionsSanityCheck) {

utilities/options/options_util_test.cc

@@ -350,6 +350,21 @@ TEST_F(OptionsUtilTest, SanityCheck) {
     ASSERT_OK(
         CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
   }
+
+  // persist_user_defined_timestamps
+  {
+    bool prev_persist_user_defined_timestamps =
+        cf_descs[2].options.persist_user_defined_timestamps;
+    cf_descs[2].options.persist_user_defined_timestamps = false;
+    ASSERT_NOK(
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
+
+    cf_descs[2].options.persist_user_defined_timestamps =
+        prev_persist_user_defined_timestamps;
+    ASSERT_OK(
+        CheckOptionsCompatibility(config_options, dbname_, db_opt, cf_descs));
+  }
+
   ASSERT_OK(DestroyDB(dbname_, Options(db_opt, cf_descs[0].options)));
 }