Customize CompressedSecondaryCache by block kind (#11204)

Summary:
Added `do_not_compress_roles` to `CompressedSecondaryCacheOptions` to disable compression on certain kinds of block. Filter blocks are now not compressed by CompressedSecondaryCache by default.

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

Test Plan: unit test added

Reviewed By: anand1976

Differential Revision: D43147698

Pulled By: pdillinger

fbshipit-source-id: db496975ae975fa18f157f93fe131a16315ac875

HISTORY.md

@@ -32,6 +32,7 @@
 
 ### New Features
 * Compaction filters are now supported for wide-column entities by means of the `FilterV3` API. See the comment of the API for more details.
+* Added `do_not_compress_roles` to `CompressedSecondaryCacheOptions` to disable compression on certain kinds of block. Filter blocks are now not compressed by CompressedSecondaryCache by default.
 * Added a new `MultiGetEntity` API that enables batched wide-column point lookups. See the API comments for more details.
 
 ## 7.10.0 (01/23/2023)

cache/compressed_secondary_cache.cc

@@ -22,12 +22,13 @@ CompressedSecondaryCache::CompressedSecondaryCache(
     std::shared_ptr<MemoryAllocator> memory_allocator, bool use_adaptive_mutex,
     CacheMetadataChargePolicy metadata_charge_policy,
     CompressionType compression_type, uint32_t compress_format_version,
-    bool enable_custom_split_merge)
+    bool enable_custom_split_merge,
+    const CacheEntryRoleSet& do_not_compress_roles)
     : cache_options_(capacity, num_shard_bits, strict_capacity_limit,
                      high_pri_pool_ratio, low_pri_pool_ratio, memory_allocator,
                      use_adaptive_mutex, metadata_charge_policy,
                      compression_type, compress_format_version,
-                     enable_custom_split_merge) {
+                     enable_custom_split_merge, do_not_compress_roles) {
   cache_ =
       NewLRUCache(capacity, num_shard_bits, strict_capacity_limit,
                   high_pri_pool_ratio, memory_allocator, use_adaptive_mutex,
@@ -71,7 +72,8 @@ std::unique_ptr<SecondaryCacheResultHandle> CompressedSecondaryCache::Lookup(
   Status s;
   Cache::ObjectPtr value{nullptr};
   size_t charge{0};
-  if (cache_options_.compression_type == kNoCompression) {
+  if (cache_options_.compression_type == kNoCompression ||
+      cache_options_.do_not_compress_roles.Contains(helper->role)) {
     s = helper->create_cb(Slice(ptr->get(), handle_value_charge),
                           create_context, allocator, &value, &charge);
   } else {
@@ -143,7 +145,8 @@ Status CompressedSecondaryCache::Insert(const Slice& key,
   Slice val(ptr.get(), size);
 
   std::string compressed_val;
-  if (cache_options_.compression_type != kNoCompression) {
+  if (cache_options_.compression_type != kNoCompression &&
+      !cache_options_.do_not_compress_roles.Contains(helper->role)) {
     PERF_COUNTER_ADD(compressed_sec_cache_uncompressed_bytes, size);
     CompressionOptions compression_opts;
     CompressionContext compression_context(cache_options_.compression_type);
@@ -314,12 +317,13 @@ std::shared_ptr<SecondaryCache> NewCompressedSecondaryCache(
     std::shared_ptr<MemoryAllocator> memory_allocator, bool use_adaptive_mutex,
     CacheMetadataChargePolicy metadata_charge_policy,
     CompressionType compression_type, uint32_t compress_format_version,
-    bool enable_custom_split_merge) {
+    bool enable_custom_split_merge,
+    const CacheEntryRoleSet& do_not_compress_roles) {
   return std::make_shared<CompressedSecondaryCache>(
       capacity, num_shard_bits, strict_capacity_limit, high_pri_pool_ratio,
       low_pri_pool_ratio, memory_allocator, use_adaptive_mutex,
       metadata_charge_policy, compression_type, compress_format_version,
-      enable_custom_split_merge);
+      enable_custom_split_merge, do_not_compress_roles);
 }
 
 std::shared_ptr<SecondaryCache> NewCompressedSecondaryCache(
@@ -331,7 +335,7 @@ std::shared_ptr<SecondaryCache> NewCompressedSecondaryCache(
       opts.high_pri_pool_ratio, opts.low_pri_pool_ratio, opts.memory_allocator,
       opts.use_adaptive_mutex, opts.metadata_charge_policy,
       opts.compression_type, opts.compress_format_version,
-      opts.enable_custom_split_merge);
+      opts.enable_custom_split_merge, opts.do_not_compress_roles);
 }
 
 }  // namespace ROCKSDB_NAMESPACE

cache/compressed_secondary_cache.h

@@ -78,7 +78,9 @@ class CompressedSecondaryCache : public SecondaryCache {
           kDefaultCacheMetadataChargePolicy,
       CompressionType compression_type = CompressionType::kLZ4Compression,
       uint32_t compress_format_version = 2,
-      bool enable_custom_split_merge = false);
+      bool enable_custom_split_merge = false,
+      const CacheEntryRoleSet& do_not_compress_roles = {
+          CacheEntryRole::kFilterBlock});
   ~CompressedSecondaryCache() override;
 
   const char* Name() const override { return "CompressedSecondaryCache"; }

cache/compressed_secondary_cache_test.cc

@@ -5,6 +5,7 @@
 
 #include "cache/compressed_secondary_cache.h"
 
+#include <array>
 #include <iterator>
 #include <memory>
 #include <tuple>
@@ -75,12 +76,23 @@ class CompressedSecondaryCacheTest : public testing::Test,
     return Status::OK();
   }
 
-  static constexpr Cache::CacheItemHelper kHelper{
+  static constexpr auto GenerateHelpersByRole() {
-      CacheEntryRole::kMisc, &DeletionCallback, &SizeCallback, &SaveToCallback,
+    std::array<Cache::CacheItemHelper, kNumCacheEntryRoles> a;
-      &CreateCallback};
+    for (uint32_t i = 0; i < kNumCacheEntryRoles; ++i) {
+      a[i] = Cache::CacheItemHelper{static_cast<CacheEntryRole>(i),
+                                    &DeletionCallback, &SizeCallback,
+                                    &SaveToCallback, &CreateCallback};
+    }
+    return a;
+  }
+
+  static const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>
+      kHelperByRole;
+
+  static const Cache::CacheItemHelper& kHelper;
 
   static constexpr Cache::CacheItemHelper kHelperFail{
-      CacheEntryRole::kMisc, &DeletionCallback, &SizeCallback,
+      CacheEntryRole::kDataBlock, &DeletionCallback, &SizeCallback,
       &SaveToCallbackFail, &CreateCallback};
 
   void SetFailCreate(bool fail) { fail_create_ = fail; }
@@ -787,6 +799,13 @@ class CompressedSecondaryCacheTest : public testing::Test,
   bool fail_create_;
 };
 
+const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>
+    CompressedSecondaryCacheTest::kHelperByRole = GenerateHelpersByRole();
+
+const Cache::CacheItemHelper& CompressedSecondaryCacheTest::kHelper =
+    CompressedSecondaryCacheTest::kHelperByRole[static_cast<int>(
+        CacheEntryRole::kDataBlock)];
+
 class CompressedSecCacheTestWithCompressAndAllocatorParam
     : public CompressedSecondaryCacheTest,
       public ::testing::WithParamInterface<std::tuple<bool, bool>> {
@@ -906,6 +925,77 @@ TEST_P(CompressedSecondaryCacheTestWithCompressionParam,
   IntegrationFullCapacityTest(sec_cache_is_compressed_);
 }
 
+TEST_P(CompressedSecondaryCacheTestWithCompressionParam, EntryRoles) {
+  CompressedSecondaryCacheOptions opts;
+  opts.capacity = 2048;
+  opts.num_shard_bits = 0;
+
+  if (sec_cache_is_compressed_) {
+    if (!LZ4_Supported()) {
+      ROCKSDB_GTEST_SKIP("This test requires LZ4 support.");
+      return;
+    }
+  } else {
+    opts.compression_type = CompressionType::kNoCompression;
+  }
+
+  // Select a random subset to include, for fast test
+  Random& r = *Random::GetTLSInstance();
+  CacheEntryRoleSet do_not_compress;
+  for (uint32_t i = 0; i < kNumCacheEntryRoles; ++i) {
+    // A few included on average, but decent chance of zero
+    if (r.OneIn(5)) {
+      do_not_compress.Add(static_cast<CacheEntryRole>(i));
+    }
+  }
+  opts.do_not_compress_roles = do_not_compress;
+
+  std::shared_ptr<SecondaryCache> sec_cache = NewCompressedSecondaryCache(opts);
+
+  // Fixed seed to ensure consistent compressibility (doesn't compress)
+  std::string junk(Random(301).RandomString(1000));
+
+  for (uint32_t i = 0; i < kNumCacheEntryRoles; ++i) {
+    CacheEntryRole role = static_cast<CacheEntryRole>(i);
+
+    // Uniquify `junk`
+    junk[0] = static_cast<char>(i);
+    TestItem item{junk.data(), junk.length()};
+    Slice ith_key = Slice(junk.data(), 5);
+
+    get_perf_context()->Reset();
+    ASSERT_OK(sec_cache->Insert(ith_key, &item, &kHelperByRole[i]));
+    ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 1U);
+
+    ASSERT_OK(sec_cache->Insert(ith_key, &item, &kHelperByRole[i]));
+    ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 1U);
+
+    bool is_in_sec_cache{true};
+    std::unique_ptr<SecondaryCacheResultHandle> handle =
+        sec_cache->Lookup(ith_key, &kHelperByRole[i], this, true,
+                          /*advise_erase=*/true, is_in_sec_cache);
+    ASSERT_NE(handle, nullptr);
+
+    // Lookup returns the right data
+    std::unique_ptr<TestItem> val =
+        std::unique_ptr<TestItem>(static_cast<TestItem*>(handle->Value()));
+    ASSERT_NE(val, nullptr);
+    ASSERT_EQ(memcmp(val->Buf(), item.Buf(), item.Size()), 0);
+
+    bool compressed =
+        sec_cache_is_compressed_ && !do_not_compress.Contains(role);
+    if (compressed) {
+      ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes,
+                1000);
+      ASSERT_EQ(get_perf_context()->compressed_sec_cache_compressed_bytes,
+                1007);
+    } else {
+      ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes, 0);
+      ASSERT_EQ(get_perf_context()->compressed_sec_cache_compressed_bytes, 0);
+    }
+  }
+}
+
 INSTANTIATE_TEST_CASE_P(CompressedSecCacheTests,
                         CompressedSecondaryCacheTestWithCompressionParam,
                         testing::Bool());

include/rocksdb/cache.h

@@ -16,6 +16,7 @@
 #include <string>
 
 #include "rocksdb/compression_type.h"
+#include "rocksdb/data_structure.h"
 #include "rocksdb/memory_allocator.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -70,6 +71,9 @@ constexpr uint32_t kNumCacheEntryRoles =
 // Obtain a hyphen-separated, lowercase name of a `CacheEntryRole`.
 const std::string& GetCacheEntryRoleName(CacheEntryRole);
 
+// A fast bit set for CacheEntryRoles
+using CacheEntryRoleSet = SmallEnumSet<CacheEntryRole, CacheEntryRole::kMisc>;
+
 // For use with `GetMapProperty()` for property
 // `DB::Properties::kBlockCacheEntryStats`. On success, the map will
 // be populated with all keys that can be obtained from these functions.
@@ -235,6 +239,10 @@ struct CompressedSecondaryCacheOptions : LRUCacheOptions {
   // into chunks so that they may better fit jemalloc bins.
   bool enable_custom_split_merge = false;
 
+  // Kinds of entries that should not be compressed, but can be stored.
+  // (Filter blocks are essentially non-compressible but others usually are.)
+  CacheEntryRoleSet do_not_compress_roles = {CacheEntryRole::kFilterBlock};
+
   CompressedSecondaryCacheOptions() {}
   CompressedSecondaryCacheOptions(
       size_t _capacity, int _num_shard_bits, bool _strict_capacity_limit,
@@ -245,14 +253,17 @@ struct CompressedSecondaryCacheOptions : LRUCacheOptions {
           kDefaultCacheMetadataChargePolicy,
       CompressionType _compression_type = CompressionType::kLZ4Compression,
       uint32_t _compress_format_version = 2,
-      bool _enable_custom_split_merge = false)
+      bool _enable_custom_split_merge = false,
+      const CacheEntryRoleSet& _do_not_compress_roles =
+          {CacheEntryRole::kFilterBlock})
       : LRUCacheOptions(_capacity, _num_shard_bits, _strict_capacity_limit,
                         _high_pri_pool_ratio, std::move(_memory_allocator),
                         _use_adaptive_mutex, _metadata_charge_policy,
                         _low_pri_pool_ratio),
         compression_type(_compression_type),
         compress_format_version(_compress_format_version),
-        enable_custom_split_merge(_enable_custom_split_merge) {}
+        enable_custom_split_merge(_enable_custom_split_merge),
+        do_not_compress_roles(_do_not_compress_roles) {}
 };
 
 // EXPERIMENTAL
@@ -267,7 +278,9 @@ extern std::shared_ptr<SecondaryCache> NewCompressedSecondaryCache(
         kDefaultCacheMetadataChargePolicy,
     CompressionType compression_type = CompressionType::kLZ4Compression,
     uint32_t compress_format_version = 2,
-    bool enable_custom_split_merge = false);
+    bool enable_custom_split_merge = false,
+    const CacheEntryRoleSet& _do_not_compress_roles = {
+        CacheEntryRole::kFilterBlock});
 
 extern std::shared_ptr<SecondaryCache> NewCompressedSecondaryCache(
     const CompressedSecondaryCacheOptions& opts);