Simplify tracking entries already in SecondaryCache (#11299)

Summary:
In preparation for factoring secondary cache support out of individual Cache implementations, we can get rid of the "in secondary cache" flag on entries through a workable hack: when an entry is promoted from secondary, it is inserted in primary using a helper that lacks secondary cache support, thus preventing re-insertion into secondary cache through existing logic.

This adds to the complexity of building CacheItemHelpers, because you always have to be able to get to an equivalent helper without secondary cache support, but that complexity is reasonably isolated within RocksDB typed_cache.h and test code.

gcc-7 seems to have problems with constexpr constructor referencing `this` so removed constexpr support on CacheItemHelper.

Also refactored some related test code to share common code / functionality.

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

Test Plan: existing tests

Reviewed By: anand1976

Differential Revision: D44101453

Pulled By: pdillinger

fbshipit-source-id: 7a59d0a3938ee40159c90c3e65d7004f6a272345

CMakeLists.txt

@@ -1279,6 +1279,7 @@ if(WITH_TESTS OR WITH_BENCHMARK_TOOLS)
   add_subdirectory(third-party/gtest-1.8.1/fused-src/gtest)
   add_library(testharness STATIC
   test_util/mock_time_env.cc
+  test_util/secondary_cache_test_util.cc
   test_util/testharness.cc)
   target_link_libraries(testharness gtest)
 endif()

TARGETS

@@ -701,6 +701,7 @@ cpp_library_wrapper(name="rocksdb_test_lib", srcs=[
         "db/db_with_timestamp_test_util.cc",
         "table/mock_table.cc",
         "test_util/mock_time_env.cc",
+        "test_util/secondary_cache_test_util.cc",
         "test_util/testharness.cc",
         "test_util/testutil.cc",
         "tools/block_cache_analyzer/block_cache_trace_analyzer.cc",

cache/cache.cc

@@ -16,6 +16,8 @@
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
+const Cache::CacheItemHelper kNoopCacheItemHelper{};
+
 static std::unordered_map<std::string, OptionTypeInfo>
     lru_cache_options_type_info = {
         {"capacity",

cache/cache_bench_tool.cc

@@ -255,12 +255,15 @@ void DeleteFn(Cache::ObjectPtr value, MemoryAllocator* /*alloc*/) {
   delete[] static_cast<char*>(value);
 }
 
+Cache::CacheItemHelper helper1_wos(CacheEntryRole::kDataBlock, DeleteFn);
 Cache::CacheItemHelper helper1(CacheEntryRole::kDataBlock, DeleteFn, SizeFn,
-                               SaveToFn, CreateFn);
+                               SaveToFn, CreateFn, &helper1_wos);
+Cache::CacheItemHelper helper2_wos(CacheEntryRole::kIndexBlock, DeleteFn);
 Cache::CacheItemHelper helper2(CacheEntryRole::kIndexBlock, DeleteFn, SizeFn,
-                               SaveToFn, CreateFn);
+                               SaveToFn, CreateFn, &helper2_wos);
+Cache::CacheItemHelper helper3_wos(CacheEntryRole::kFilterBlock, DeleteFn);
 Cache::CacheItemHelper helper3(CacheEntryRole::kFilterBlock, DeleteFn, SizeFn,
-                               SaveToFn, CreateFn);
+                               SaveToFn, CreateFn, &helper3_wos);
 }  // namespace
 
 class CacheBench {

cache/cache_entry_stats.h

@@ -143,7 +143,7 @@ class CacheEntryStatsCollector {
       }
     }
     // If we reach here, shared entry is in cache with handle `h`.
-    assert(cache.get()->GetCacheItemHelper(h) == &cache.kBasicHelper);
+    assert(cache.get()->GetCacheItemHelper(h) == cache.GetBasicHelper());
 
     // Build an aliasing shared_ptr that keeps `ptr` in cache while there
     // are references.

cache/cache_reservation_manager.cc

@@ -169,7 +169,7 @@ Slice CacheReservationManagerImpl<R>::GetNextCacheKey() {
 template <CacheEntryRole R>
 const Cache::CacheItemHelper*
 CacheReservationManagerImpl<R>::TEST_GetCacheItemHelperForRole() {
-  return &CacheInterface::kHelper;
+  return CacheInterface::GetHelper();
 }
 
 template class CacheReservationManagerImpl<

cache/cache_test.cc

@@ -95,8 +95,7 @@ class CacheTest : public testing::TestWithParam<std::string> {
   static void Deleter(Cache::ObjectPtr v, MemoryAllocator*) {
     current_->deleted_values_.push_back(DecodeValue(v));
   }
-  static constexpr Cache::CacheItemHelper kHelper{CacheEntryRole::kMisc,
-                                                  &Deleter};
+  static const Cache::CacheItemHelper kHelper;
 
   static const int kCacheSize = 1000;
   static const int kNumShardBits = 4;
@@ -212,6 +211,9 @@ class CacheTest : public testing::TestWithParam<std::string> {
   void Erase2(int key) { Erase(cache2_, key); }
 };
 
+const Cache::CacheItemHelper CacheTest::kHelper{CacheEntryRole::kMisc,
+                                                &CacheTest::Deleter};
+
 CacheTest* CacheTest::current_;
 std::string CacheTest::type_;
 

cache/compressed_secondary_cache_test.cc

@@ -12,98 +12,32 @@
 
 #include "memory/jemalloc_nodump_allocator.h"
 #include "rocksdb/convenience.h"
+#include "test_util/secondary_cache_test_util.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 
 namespace ROCKSDB_NAMESPACE {
 
+using secondary_cache_test_util::GetHelper;
+using secondary_cache_test_util::GetHelperFail;
+using secondary_cache_test_util::TestCreateContext;
+using secondary_cache_test_util::TestItem;
+
 class CompressedSecondaryCacheTest : public testing::Test,
-                                     public Cache::CreateContext {
+                                     public TestCreateContext {
  public:
-  CompressedSecondaryCacheTest() : fail_create_(false) {}
+  CompressedSecondaryCacheTest() {}
   ~CompressedSecondaryCacheTest() override = default;
 
  protected:
-  class TestItem {
-   public:
-    TestItem(const char* buf, size_t size) : buf_(new char[size]), size_(size) {
-      memcpy(buf_.get(), buf, size);
-    }
-    ~TestItem() = default;
-
-    char* Buf() { return buf_.get(); }
-    [[nodiscard]] size_t Size() const { return size_; }
-
-   private:
-    std::unique_ptr<char[]> buf_;
-    size_t size_;
-  };
-
-  static size_t SizeCallback(Cache::ObjectPtr obj) {
-    return static_cast<TestItem*>(obj)->Size();
-  }
-
-  static Status SaveToCallback(Cache::ObjectPtr from_obj, size_t from_offset,
-                               size_t length, char* out) {
-    auto item = static_cast<TestItem*>(from_obj);
-    const char* buf = item->Buf();
-    EXPECT_EQ(length, item->Size());
-    EXPECT_EQ(from_offset, 0);
-    memcpy(out, buf, length);
-    return Status::OK();
-  }
-
-  static void DeletionCallback(Cache::ObjectPtr obj,
-                               MemoryAllocator* /*alloc*/) {
-    delete static_cast<TestItem*>(obj);
-    obj = nullptr;
-  }
-
-  static Status SaveToCallbackFail(Cache::ObjectPtr /*obj*/, size_t /*offset*/,
-                                   size_t /*size*/, char* /*out*/) {
-    return Status::NotSupported();
-  }
-
-  static Status CreateCallback(const Slice& data, Cache::CreateContext* context,
-                               MemoryAllocator* /*allocator*/,
-                               Cache::ObjectPtr* out_obj, size_t* out_charge) {
-    auto t = static_cast<CompressedSecondaryCacheTest*>(context);
-    if (t->fail_create_) {
-      return Status::NotSupported();
-    }
-    *out_obj = new TestItem(data.data(), data.size());
-    *out_charge = data.size();
-    return Status::OK();
-  }
-
-  static constexpr auto GenerateHelpersByRole() {
-    std::array<Cache::CacheItemHelper, kNumCacheEntryRoles> a;
-    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::kDataBlock, &DeletionCallback, &SizeCallback,
-      &SaveToCallbackFail, &CreateCallback};
-
-  void SetFailCreate(bool fail) { fail_create_ = fail; }
-
   void BasicTestHelper(std::shared_ptr<SecondaryCache> sec_cache,
                        bool sec_cache_is_compressed) {
     get_perf_context()->Reset();
     bool kept_in_sec_cache{true};
     // Lookup an non-existent key.
-    std::unique_ptr<SecondaryCacheResultHandle> handle0 = sec_cache->Lookup(
-        "k0", &kHelper, this, true, /*advise_erase=*/true, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle0 =
+        sec_cache->Lookup("k0", GetHelper(), this, true, /*advise_erase=*/true,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle0, nullptr);
 
     Random rnd(301);
@@ -111,21 +45,23 @@ class CompressedSecondaryCacheTest : public testing::Test,
     std::string str1(rnd.RandomString(1000));
     TestItem item1(str1.data(), str1.length());
     // A dummy handle is inserted if the item is inserted for the first time.
-    ASSERT_OK(sec_cache->Insert("k1", &item1, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k1", &item1, GetHelper()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 1);
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes, 0);
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_compressed_bytes, 0);
 
-    std::unique_ptr<SecondaryCacheResultHandle> handle1_1 = sec_cache->Lookup(
-        "k1", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle1_1 =
+        sec_cache->Lookup("k1", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle1_1, nullptr);
 
     // Insert and Lookup the item k1 for the second time and advise erasing it.
-    ASSERT_OK(sec_cache->Insert("k1", &item1, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k1", &item1, GetHelper()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 1);
 
-    std::unique_ptr<SecondaryCacheResultHandle> handle1_2 = sec_cache->Lookup(
-        "k1", &kHelper, this, true, /*advise_erase=*/true, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle1_2 =
+        sec_cache->Lookup("k1", GetHelper(), this, true, /*advise_erase=*/true,
+                          kept_in_sec_cache);
     ASSERT_NE(handle1_2, nullptr);
     ASSERT_FALSE(kept_in_sec_cache);
     if (sec_cache_is_compressed) {
@@ -144,20 +80,22 @@ class CompressedSecondaryCacheTest : public testing::Test,
     ASSERT_EQ(memcmp(val1->Buf(), item1.Buf(), item1.Size()), 0);
 
     // Lookup the item k1 again.
-    std::unique_ptr<SecondaryCacheResultHandle> handle1_3 = sec_cache->Lookup(
-        "k1", &kHelper, this, true, /*advise_erase=*/true, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle1_3 =
+        sec_cache->Lookup("k1", GetHelper(), this, true, /*advise_erase=*/true,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle1_3, nullptr);
 
     // Insert and Lookup the item k2.
     std::string str2(rnd.RandomString(1000));
     TestItem item2(str2.data(), str2.length());
-    ASSERT_OK(sec_cache->Insert("k2", &item2, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k2", &item2, GetHelper()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 2);
-    std::unique_ptr<SecondaryCacheResultHandle> handle2_1 = sec_cache->Lookup(
-        "k2", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle2_1 =
+        sec_cache->Lookup("k2", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle2_1, nullptr);
 
-    ASSERT_OK(sec_cache->Insert("k2", &item2, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k2", &item2, GetHelper()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 2);
     if (sec_cache_is_compressed) {
       ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes,
@@ -168,8 +106,9 @@ class CompressedSecondaryCacheTest : public testing::Test,
       ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes, 0);
       ASSERT_EQ(get_perf_context()->compressed_sec_cache_compressed_bytes, 0);
     }
-    std::unique_ptr<SecondaryCacheResultHandle> handle2_2 = sec_cache->Lookup(
-        "k2", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle2_2 =
+        sec_cache->Lookup("k2", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_NE(handle2_2, nullptr);
     std::unique_ptr<TestItem> val2 =
         std::unique_ptr<TestItem>(static_cast<TestItem*>(handle2_2->Value()));
@@ -238,24 +177,26 @@ class CompressedSecondaryCacheTest : public testing::Test,
     std::string str1(rnd.RandomString(1000));
     TestItem item1(str1.data(), str1.length());
     // Insert a dummy handle.
-    ASSERT_OK(sec_cache->Insert("k1", &item1, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k1", &item1, GetHelper()));
     // Insert k1.
-    ASSERT_OK(sec_cache->Insert("k1", &item1, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k1", &item1, GetHelper()));
 
     // Insert and Lookup the second item.
     std::string str2(rnd.RandomString(200));
     TestItem item2(str2.data(), str2.length());
     // Insert a dummy handle, k1 is not evicted.
-    ASSERT_OK(sec_cache->Insert("k2", &item2, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k2", &item2, GetHelper()));
     bool kept_in_sec_cache{false};
-    std::unique_ptr<SecondaryCacheResultHandle> handle1 = sec_cache->Lookup(
-        "k1", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle1 =
+        sec_cache->Lookup("k1", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle1, nullptr);
 
     // Insert k2 and k1 is evicted.
-    ASSERT_OK(sec_cache->Insert("k2", &item2, &kHelper));
-    std::unique_ptr<SecondaryCacheResultHandle> handle2 = sec_cache->Lookup(
-        "k2", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    ASSERT_OK(sec_cache->Insert("k2", &item2, GetHelper()));
+    std::unique_ptr<SecondaryCacheResultHandle> handle2 =
+        sec_cache->Lookup("k2", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_NE(handle2, nullptr);
     std::unique_ptr<TestItem> val2 =
         std::unique_ptr<TestItem>(static_cast<TestItem*>(handle2->Value()));
@@ -263,24 +204,26 @@ class CompressedSecondaryCacheTest : public testing::Test,
     ASSERT_EQ(memcmp(val2->Buf(), item2.Buf(), item2.Size()), 0);
 
     // Insert k1 again and a dummy handle is inserted.
-    ASSERT_OK(sec_cache->Insert("k1", &item1, &kHelper));
+    ASSERT_OK(sec_cache->Insert("k1", &item1, GetHelper()));
 
-    std::unique_ptr<SecondaryCacheResultHandle> handle1_1 = sec_cache->Lookup(
-        "k1", &kHelper, this, true, /*advise_erase=*/false, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle1_1 =
+        sec_cache->Lookup("k1", GetHelper(), this, true, /*advise_erase=*/false,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle1_1, nullptr);
 
     // Create Fails.
     SetFailCreate(true);
-    std::unique_ptr<SecondaryCacheResultHandle> handle2_1 = sec_cache->Lookup(
-        "k2", &kHelper, this, true, /*advise_erase=*/true, kept_in_sec_cache);
+    std::unique_ptr<SecondaryCacheResultHandle> handle2_1 =
+        sec_cache->Lookup("k2", GetHelper(), this, true, /*advise_erase=*/true,
+                          kept_in_sec_cache);
     ASSERT_EQ(handle2_1, nullptr);
 
     // Save Fails.
     std::string str3 = rnd.RandomString(10);
     TestItem item3(str3.data(), str3.length());
     // The Status is OK because a dummy handle is inserted.
-    ASSERT_OK(sec_cache->Insert("k3", &item3, &kHelperFail));
-    ASSERT_NOK(sec_cache->Insert("k3", &item3, &kHelperFail));
+    ASSERT_OK(sec_cache->Insert("k3", &item3, GetHelperFail()));
+    ASSERT_NOK(sec_cache->Insert("k3", &item3, GetHelperFail()));
 
     sec_cache.reset();
   }
@@ -317,13 +260,13 @@ class CompressedSecondaryCacheTest : public testing::Test,
     Random rnd(301);
     std::string str1 = rnd.RandomString(1001);
     auto item1_1 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1_1, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1_1, GetHelper(), str1.length()));
 
     std::string str2 = rnd.RandomString(1012);
     auto item2_1 = new TestItem(str2.data(), str2.length());
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's dummy item.
-    ASSERT_OK(cache->Insert("k2", item2_1, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2_1, GetHelper(), str2.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 1);
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes, 0);
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_compressed_bytes, 0);
@@ -332,19 +275,19 @@ class CompressedSecondaryCacheTest : public testing::Test,
     auto item3_1 = new TestItem(str3.data(), str3.length());
     // After this Insert, primary cache contains k3 and secondary cache contains
     // k1's dummy item and k2's dummy item.
-    ASSERT_OK(cache->Insert("k3", item3_1, &kHelper, str3.length()));
+    ASSERT_OK(cache->Insert("k3", item3_1, GetHelper(), str3.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 2);
 
     // After this Insert, primary cache contains k1 and secondary cache contains
     // k1's dummy item, k2's dummy item, and k3's dummy item.
     auto item1_2 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1_2, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1_2, GetHelper(), str1.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 3);
 
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's item, k2's dummy item, and k3's dummy item.
     auto item2_2 = new TestItem(str2.data(), str2.length());
-    ASSERT_OK(cache->Insert("k2", item2_2, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2_2, GetHelper(), str2.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 1);
     if (sec_cache_is_compressed) {
       ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes,
@@ -359,7 +302,7 @@ class CompressedSecondaryCacheTest : public testing::Test,
     // After this Insert, primary cache contains k3 and secondary cache contains
     // k1's item and k2's item.
     auto item3_2 = new TestItem(str3.data(), str3.length());
-    ASSERT_OK(cache->Insert("k3", item3_2, &kHelper, str3.length()));
+    ASSERT_OK(cache->Insert("k3", item3_2, GetHelper(), str3.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 2);
     if (sec_cache_is_compressed) {
       ASSERT_EQ(get_perf_context()->compressed_sec_cache_uncompressed_bytes,
@@ -372,7 +315,7 @@ class CompressedSecondaryCacheTest : public testing::Test,
     }
 
     Cache::Handle* handle;
-    handle = cache->Lookup("k3", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k3", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_NE(handle, nullptr);
     auto val3 = static_cast<TestItem*>(cache->Value(handle));
@@ -381,13 +324,13 @@ class CompressedSecondaryCacheTest : public testing::Test,
     cache->Release(handle);
 
     // Lookup an non-existent key.
-    handle = cache->Lookup("k0", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k0", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_EQ(handle, nullptr);
 
     // This Lookup should just insert a dummy handle in the primary cache
     // and the k1 is still in the secondary cache.
-    handle = cache->Lookup("k1", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k1", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_NE(handle, nullptr);
     ASSERT_EQ(get_perf_context()->block_cache_standalone_handle_count, 1);
@@ -399,7 +342,7 @@ class CompressedSecondaryCacheTest : public testing::Test,
     // This Lookup should erase k1 from the secondary cache and insert
     // it into primary cache; then k3 is demoted.
     // k2 and k3 are in secondary cache.
-    handle = cache->Lookup("k1", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k1", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_NE(handle, nullptr);
     ASSERT_EQ(get_perf_context()->block_cache_standalone_handle_count, 1);
@@ -407,7 +350,7 @@ class CompressedSecondaryCacheTest : public testing::Test,
     cache->Release(handle);
 
     // k2 is still in secondary cache.
-    handle = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_NE(handle, nullptr);
     ASSERT_EQ(get_perf_context()->block_cache_standalone_handle_count, 2);
@@ -415,7 +358,7 @@ class CompressedSecondaryCacheTest : public testing::Test,
 
     // Testing SetCapacity().
     ASSERT_OK(secondary_cache->SetCapacity(0));
-    handle = cache->Lookup("k3", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k3", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
     ASSERT_EQ(handle, nullptr);
 
@@ -425,30 +368,30 @@ class CompressedSecondaryCacheTest : public testing::Test,
     ASSERT_EQ(capacity, 7000);
     auto item1_3 = new TestItem(str1.data(), str1.length());
     // After this Insert, primary cache contains k1.
-    ASSERT_OK(cache->Insert("k1", item1_3, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k1", item1_3, GetHelper(), str2.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 3);
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 4);
 
     auto item2_3 = new TestItem(str2.data(), str2.length());
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's dummy item.
-    ASSERT_OK(cache->Insert("k2", item2_3, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k2", item2_3, GetHelper(), str1.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 4);
 
     auto item1_4 = new TestItem(str1.data(), str1.length());
     // After this Insert, primary cache contains k1 and secondary cache contains
     // k1's dummy item and k2's dummy item.
-    ASSERT_OK(cache->Insert("k1", item1_4, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k1", item1_4, GetHelper(), str2.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 5);
 
     auto item2_4 = new TestItem(str2.data(), str2.length());
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's real item and k2's dummy item.
-    ASSERT_OK(cache->Insert("k2", item2_4, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2_4, GetHelper(), str2.length()));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 5);
     // This Lookup should just insert a dummy handle in the primary cache
     // and the k1 is still in the secondary cache.
-    handle = cache->Lookup("k1", &kHelper, this, Cache::Priority::LOW, true,
+    handle = cache->Lookup("k1", GetHelper(), this, Cache::Priority::LOW, true,
                            stats.get());
 
     ASSERT_NE(handle, nullptr);
@@ -487,13 +430,14 @@ class CompressedSecondaryCacheTest : public testing::Test,
     Random rnd(301);
     std::string str1 = rnd.RandomString(1001);
     auto item1 = std::make_unique<TestItem>(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1.get(), &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1.get(), GetHelper(), str1.length()));
     item1.release();  // Appease clang-analyze "potential memory leak"
 
     Cache::Handle* handle;
     handle = cache->Lookup("k2", nullptr, this, Cache::Priority::LOW, true);
     ASSERT_EQ(handle, nullptr);
-    handle = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, false);
+    handle =
+        cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, false);
     ASSERT_EQ(handle, nullptr);
 
     cache.reset();
@@ -529,25 +473,25 @@ class CompressedSecondaryCacheTest : public testing::Test,
     Random rnd(301);
     std::string str1 = rnd.RandomString(1001);
     auto item1 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1, &kHelperFail, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1, GetHelperFail(), str1.length()));
 
     std::string str2 = rnd.RandomString(1002);
     auto item2 = new TestItem(str2.data(), str2.length());
     // k1 should be demoted to the secondary cache.
-    ASSERT_OK(cache->Insert("k2", item2, &kHelperFail, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2, GetHelperFail(), str2.length()));
 
     Cache::Handle* handle;
     handle =
-        cache->Lookup("k2", &kHelperFail, this, Cache::Priority::LOW, true);
+        cache->Lookup("k2", GetHelperFail(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle, nullptr);
     cache->Release(handle);
     // This lookup should fail, since k1 demotion would have failed.
     handle =
-        cache->Lookup("k1", &kHelperFail, this, Cache::Priority::LOW, true);
+        cache->Lookup("k1", GetHelperFail(), this, Cache::Priority::LOW, true);
     ASSERT_EQ(handle, nullptr);
     // Since k1 was not promoted, k2 should still be in cache.
     handle =
-        cache->Lookup("k2", &kHelperFail, this, Cache::Priority::LOW, true);
+        cache->Lookup("k2", GetHelperFail(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle, nullptr);
     cache->Release(handle);
 
@@ -584,23 +528,23 @@ class CompressedSecondaryCacheTest : public testing::Test,
     Random rnd(301);
     std::string str1 = rnd.RandomString(1001);
     auto item1 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1, GetHelper(), str1.length()));
 
     std::string str2 = rnd.RandomString(1002);
     auto item2 = new TestItem(str2.data(), str2.length());
     // k1 should be demoted to the secondary cache.
-    ASSERT_OK(cache->Insert("k2", item2, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2, GetHelper(), str2.length()));
 
     Cache::Handle* handle;
     SetFailCreate(true);
-    handle = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, true);
+    handle = cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle, nullptr);
     cache->Release(handle);
     // This lookup should fail, since k1 creation would have failed
-    handle = cache->Lookup("k1", &kHelper, this, Cache::Priority::LOW, true);
+    handle = cache->Lookup("k1", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_EQ(handle, nullptr);
     // Since k1 didn't get promoted, k2 should still be in cache
-    handle = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, true);
+    handle = cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle, nullptr);
     cache->Release(handle);
 
@@ -637,27 +581,28 @@ class CompressedSecondaryCacheTest : public testing::Test,
     Random rnd(301);
     std::string str1 = rnd.RandomString(1001);
     auto item1_1 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1_1, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1_1, GetHelper(), str1.length()));
 
     std::string str2 = rnd.RandomString(1002);
     std::string str2_clone{str2};
     auto item2 = new TestItem(str2.data(), str2.length());
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's dummy item.
-    ASSERT_OK(cache->Insert("k2", item2, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2, GetHelper(), str2.length()));
 
     // After this Insert, primary cache contains k1 and secondary cache contains
     // k1's dummy item and k2's dummy item.
     auto item1_2 = new TestItem(str1.data(), str1.length());
-    ASSERT_OK(cache->Insert("k1", item1_2, &kHelper, str1.length()));
+    ASSERT_OK(cache->Insert("k1", item1_2, GetHelper(), str1.length()));
 
     auto item2_2 = new TestItem(str2.data(), str2.length());
     // After this Insert, primary cache contains k2 and secondary cache contains
     // k1's item and k2's dummy item.
-    ASSERT_OK(cache->Insert("k2", item2_2, &kHelper, str2.length()));
+    ASSERT_OK(cache->Insert("k2", item2_2, GetHelper(), str2.length()));
 
     Cache::Handle* handle2;
-    handle2 = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, true);
+    handle2 =
+        cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle2, nullptr);
     cache->Release(handle2);
 
@@ -665,12 +610,14 @@ class CompressedSecondaryCacheTest : public testing::Test,
     // strict_capacity_limit is true, but the lookup should still succeed.
     // A k1's dummy item is inserted into primary cache.
     Cache::Handle* handle1;
-    handle1 = cache->Lookup("k1", &kHelper, this, Cache::Priority::LOW, true);
+    handle1 =
+        cache->Lookup("k1", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle1, nullptr);
     cache->Release(handle1);
 
     // Since k1 didn't get inserted, k2 should still be in cache
-    handle2 = cache->Lookup("k2", &kHelper, this, Cache::Priority::LOW, true);
+    handle2 =
+        cache->Lookup("k2", GetHelper(), this, Cache::Priority::LOW, true);
     ASSERT_NE(handle2, nullptr);
     cache->Release(handle2);
 
@@ -794,18 +741,8 @@ class CompressedSecondaryCacheTest : public testing::Test,
 
     sec_cache->GetHelper(true)->del_cb(chunks_head, /*alloc*/ nullptr);
   }
-
- private:
-  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>> {
@@ -964,15 +901,15 @@ TEST_P(CompressedSecondaryCacheTestWithCompressionParam, EntryRoles) {
     Slice ith_key = Slice(junk.data(), 5);
 
     get_perf_context()->Reset();
-    ASSERT_OK(sec_cache->Insert(ith_key, &item, &kHelperByRole[i]));
+    ASSERT_OK(sec_cache->Insert(ith_key, &item, GetHelper(role)));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_dummy_count, 1U);
 
-    ASSERT_OK(sec_cache->Insert(ith_key, &item, &kHelperByRole[i]));
+    ASSERT_OK(sec_cache->Insert(ith_key, &item, GetHelper(role)));
     ASSERT_EQ(get_perf_context()->compressed_sec_cache_insert_real_count, 1U);
 
     bool kept_in_sec_cache{true};
     std::unique_ptr<SecondaryCacheResultHandle> handle =
-        sec_cache->Lookup(ith_key, &kHelperByRole[i], this, true,
+        sec_cache->Lookup(ith_key, GetHelper(role), this, true,
                           /*advise_erase=*/true, kept_in_sec_cache);
     ASSERT_NE(handle, nullptr);
 

cache/lru_cache.cc

@@ -344,8 +344,7 @@ void LRUCacheShard::EvictFromLRU(size_t charge,
 void LRUCacheShard::TryInsertIntoSecondaryCache(
     autovector<LRUHandle*> evicted_handles) {
   for (auto entry : evicted_handles) {
-    if (secondary_cache_ && entry->IsSecondaryCacheCompatible() &&
-        !entry->IsInSecondaryCache()) {
+    if (secondary_cache_ && entry->IsSecondaryCacheCompatible()) {
       secondary_cache_->Insert(entry->key(), entry->value, entry->helper)
           .PermitUncheckedError();
     }
@@ -562,6 +561,12 @@ LRUHandle* LRUCacheShard::Lookup(const Slice& key, uint32_t hash,
     if (secondary_handle != nullptr) {
       e = static_cast<LRUHandle*>(malloc(sizeof(LRUHandle) - 1 + key.size()));
 
+      // For entries already in secondary cache, prevent re-insertion by
+      // using a helper that is not secondary cache compatible
+      if (kept_in_sec_cache) {
+        helper = helper->without_secondary_compat;
+      }
+
       e->m_flags = 0;
       e->im_flags = 0;
       e->helper = helper;
@@ -575,7 +580,6 @@ LRUHandle* LRUCacheShard::Lookup(const Slice& key, uint32_t hash,
       e->sec_handle = secondary_handle.release();
       e->total_charge = 0;
       e->Ref();
-      e->SetIsInSecondaryCache(kept_in_sec_cache);
       e->SetIsStandalone(secondary_cache_->SupportForceErase() &&
                          !found_dummy_entry);
 

cache/lru_cache.h

@@ -91,10 +91,8 @@ struct LRUHandle {
     IM_IS_LOW_PRI = (1 << 1),
     // Is the handle still being read from a lower tier.
     IM_IS_PENDING = (1 << 2),
-    // Whether this handle is still in a lower tier
-    IM_IS_IN_SECONDARY_CACHE = (1 << 3),
     // Marks result handles that should not be inserted into cache
-    IM_IS_STANDALONE = (1 << 4),
+    IM_IS_STANDALONE = (1 << 3),
   };
 
   // Beginning of the key (MUST BE THE LAST FIELD IN THIS STRUCT!)
@@ -126,9 +124,6 @@ struct LRUHandle {
   bool HasHit() const { return m_flags & M_HAS_HIT; }
   bool IsSecondaryCacheCompatible() const { return helper->size_cb != nullptr; }
   bool IsPending() const { return im_flags & IM_IS_PENDING; }
-  bool IsInSecondaryCache() const {
-    return im_flags & IM_IS_IN_SECONDARY_CACHE;
-  }
   bool IsStandalone() const { return im_flags & IM_IS_STANDALONE; }
 
   void SetInCache(bool in_cache) {
@@ -178,14 +173,6 @@ struct LRUHandle {
     }
   }
 
-  void SetIsInSecondaryCache(bool is_in_secondary_cache) {
-    if (is_in_secondary_cache) {
-      im_flags |= IM_IS_IN_SECONDARY_CACHE;
-    } else {
-      im_flags &= ~IM_IS_IN_SECONDARY_CACHE;
-    }
-  }
-
   void SetIsStandalone(bool is_standalone) {
     if (is_standalone) {
       im_flags |= IM_IS_STANDALONE;

cache/lru_cache_test.cc

@@ -19,6 +19,7 @@
 #include "rocksdb/io_status.h"
 #include "rocksdb/sst_file_manager.h"
 #include "rocksdb/utilities/cache_dump_load.h"
+#include "test_util/secondary_cache_test_util.h"
 #include "test_util/testharness.h"
 #include "typed_cache.h"
 #include "util/coding.h"
@@ -1052,6 +1053,11 @@ class TestSecondaryCache : public SecondaryCache {
   ResultMap result_map_;
 };
 
+using secondary_cache_test_util::GetHelper;
+using secondary_cache_test_util::GetHelperFail;
+using secondary_cache_test_util::TestCreateContext;
+using secondary_cache_test_util::TestItem;
+
 class DBSecondaryCacheTest : public DBTestBase {
  public:
   DBSecondaryCacheTest()
@@ -1065,86 +1071,7 @@ class DBSecondaryCacheTest : public DBTestBase {
 };
 
 class LRUCacheSecondaryCacheTest : public LRUCacheTest,
-                                   public Cache::CreateContext {
- public:
-  LRUCacheSecondaryCacheTest() : fail_create_(false) {}
-  ~LRUCacheSecondaryCacheTest() {}
-
- protected:
-  class TestItem {
-   public:
-    TestItem(const char* buf, size_t size) : buf_(new char[size]), size_(size) {
-      memcpy(buf_.get(), buf, size);
-    }
-    ~TestItem() {}
-
-    char* Buf() { return buf_.get(); }
-    size_t Size() { return size_; }
-    std::string ToString() { return std::string(Buf(), Size()); }
-
-   private:
-    std::unique_ptr<char[]> buf_;
-    size_t size_;
-  };
-
-  static size_t SizeCallback(Cache::ObjectPtr obj) {
-    return static_cast<TestItem*>(obj)->Size();
-  }
-
-  static Status SaveToCallback(Cache::ObjectPtr from_obj, size_t from_offset,
-                               size_t length, char* out) {
-    TestItem* item = static_cast<TestItem*>(from_obj);
-    char* buf = item->Buf();
-    EXPECT_EQ(length, item->Size());
-    EXPECT_EQ(from_offset, 0);
-    memcpy(out, buf, length);
-    return Status::OK();
-  }
-
-  static void DeletionCallback(Cache::ObjectPtr obj,
-                               MemoryAllocator* /*alloc*/) {
-    delete static_cast<TestItem*>(obj);
-  }
-
-  static Cache::CacheItemHelper helper_;
-
-  static Status SaveToCallbackFail(Cache::ObjectPtr /*from_obj*/,
-                                   size_t /*from_offset*/, size_t /*length*/,
-                                   char* /*out*/) {
-    return Status::NotSupported();
-  }
-
-  static Cache::CacheItemHelper helper_fail_;
-
-  static Status CreateCallback(const Slice& data, Cache::CreateContext* context,
-                               MemoryAllocator* /*allocator*/,
-                               Cache::ObjectPtr* out_obj, size_t* out_charge) {
-    auto t = static_cast<LRUCacheSecondaryCacheTest*>(context);
-    if (t->fail_create_) {
-      return Status::NotSupported();
-    }
-    *out_obj = new TestItem(data.data(), data.size());
-    *out_charge = data.size();
-    return Status::OK();
-  };
-
-  void SetFailCreate(bool fail) { fail_create_ = fail; }
-
- private:
-  bool fail_create_;
-};
-
-Cache::CacheItemHelper LRUCacheSecondaryCacheTest::helper_{
-    CacheEntryRole::kMisc, LRUCacheSecondaryCacheTest::DeletionCallback,
-    LRUCacheSecondaryCacheTest::SizeCallback,
-    LRUCacheSecondaryCacheTest::SaveToCallback,
-    LRUCacheSecondaryCacheTest::CreateCallback};
-
-Cache::CacheItemHelper LRUCacheSecondaryCacheTest::helper_fail_{
-    CacheEntryRole::kMisc, LRUCacheSecondaryCacheTest::DeletionCallback,
-    LRUCacheSecondaryCacheTest::SizeCallback,
-    LRUCacheSecondaryCacheTest::SaveToCallbackFail,
-    LRUCacheSecondaryCacheTest::CreateCallback};
+                                   public TestCreateContext {};
 
 TEST_F(LRUCacheSecondaryCacheTest, BasicTest) {
   LRUCacheOptions opts(1024 /* capacity */, 0 /* num_shard_bits */,
@@ -1168,18 +1095,16 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicTest) {
 
   std::string str1 = rnd.RandomString(1020);
   TestItem* item1 = new TestItem(str1.data(), str1.length());
-  ASSERT_OK(cache->Insert(k1.AsSlice(), item1,
-                          &LRUCacheSecondaryCacheTest::helper_, str1.length()));
+  ASSERT_OK(cache->Insert(k1.AsSlice(), item1, GetHelper(), str1.length()));
   std::string str2 = rnd.RandomString(1021);
   TestItem* item2 = new TestItem(str2.data(), str2.length());
   // k1 should be demoted to NVM
-  ASSERT_OK(cache->Insert(k2.AsSlice(), item2,
-                          &LRUCacheSecondaryCacheTest::helper_, str2.length()));
+  ASSERT_OK(cache->Insert(k2.AsSlice(), item2, GetHelper(), str2.length()));
 
   get_perf_context()->Reset();
   Cache::Handle* handle;
   handle =
-      cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+      cache->Lookup(k2.AsSlice(), GetHelper(),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str2.size());
@@ -1187,7 +1112,7 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicTest) {
 
   // This lookup should promote k1 and demote k2
   handle =
-      cache->Lookup(k1.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+      cache->Lookup(k1.AsSlice(), GetHelper(),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str1.size());
@@ -1195,7 +1120,7 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicTest) {
 
   // This lookup should promote k3 and demote k1
   handle =
-      cache->Lookup(k3.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+      cache->Lookup(k3.AsSlice(), GetHelper(),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str3.size());
@@ -1226,12 +1151,6 @@ TEST_F(LRUCacheSecondaryCacheTest, StatsTest) {
   CacheKey k1 = CacheKey::CreateUniqueForCacheLifetime(cache.get());
   CacheKey k2 = CacheKey::CreateUniqueForCacheLifetime(cache.get());
   CacheKey k3 = CacheKey::CreateUniqueForCacheLifetime(cache.get());
-  Cache::CacheItemHelper filter_helper = helper_;
-  Cache::CacheItemHelper index_helper = helper_;
-  Cache::CacheItemHelper data_helper = helper_;
-  filter_helper.role = CacheEntryRole::kFilterBlock;
-  index_helper.role = CacheEntryRole::kIndexBlock;
-  data_helper.role = CacheEntryRole::kDataBlock;
 
   Random rnd(301);
   // Start with warming secondary cache
@@ -1245,21 +1164,21 @@ TEST_F(LRUCacheSecondaryCacheTest, StatsTest) {
   get_perf_context()->Reset();
   Cache::Handle* handle;
   handle =
-      cache->Lookup(k1.AsSlice(), &filter_helper,
+      cache->Lookup(k1.AsSlice(), GetHelper(CacheEntryRole::kFilterBlock),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str1.size());
   cache->Release(handle);
 
   handle =
-      cache->Lookup(k2.AsSlice(), &index_helper,
+      cache->Lookup(k2.AsSlice(), GetHelper(CacheEntryRole::kIndexBlock),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str2.size());
   cache->Release(handle);
 
   handle =
-      cache->Lookup(k3.AsSlice(), &data_helper,
+      cache->Lookup(k3.AsSlice(), GetHelper(CacheEntryRole::kDataBlock),
                     /*context*/ this, Cache::Priority::LOW, true, stats.get());
   ASSERT_NE(handle, nullptr);
   ASSERT_EQ(static_cast<TestItem*>(cache->Value(handle))->Size(), str3.size());
@@ -1298,15 +1217,15 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicFailTest) {
   // NOTE: changed to assert helper != nullptr for efficiency / code size
   // ASSERT_TRUE(cache->Insert(k1.AsSlice(), item1.get(), nullptr,
   //                           str1.length()).IsInvalidArgument());
-  ASSERT_OK(cache->Insert(k1.AsSlice(), item1.get(),
-                          &LRUCacheSecondaryCacheTest::helper_, str1.length()));
+  ASSERT_OK(
+      cache->Insert(k1.AsSlice(), item1.get(), GetHelper(), str1.length()));
   item1.release();  // Appease clang-analyze "potential memory leak"
 
   Cache::Handle* handle;
   handle = cache->Lookup(k2.AsSlice(), nullptr, /*context*/ this,
                          Cache::Priority::LOW, true);
   ASSERT_EQ(handle, nullptr);
-  handle = cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, false);
   ASSERT_EQ(handle, nullptr);
 
@@ -1330,32 +1249,25 @@ TEST_F(LRUCacheSecondaryCacheTest, SaveFailTest) {
   Random rnd(301);
   std::string str1 = rnd.RandomString(1020);
   TestItem* item1 = new TestItem(str1.data(), str1.length());
-  ASSERT_OK(cache->Insert(k1.AsSlice(), item1,
-                          &LRUCacheSecondaryCacheTest::helper_fail_,
-                          str1.length()));
+  ASSERT_OK(cache->Insert(k1.AsSlice(), item1, GetHelperFail(), str1.length()));
   std::string str2 = rnd.RandomString(1020);
   TestItem* item2 = new TestItem(str2.data(), str2.length());
   // k1 should be demoted to NVM
   ASSERT_EQ(secondary_cache->num_inserts(), 0u);
-  ASSERT_OK(cache->Insert(k2.AsSlice(), item2,
-                          &LRUCacheSecondaryCacheTest::helper_fail_,
-                          str2.length()));
+  ASSERT_OK(cache->Insert(k2.AsSlice(), item2, GetHelperFail(), str2.length()));
   ASSERT_EQ(secondary_cache->num_inserts(), 1u);
 
   Cache::Handle* handle;
-  handle =
-      cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_fail_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelperFail(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   cache->Release(handle);
   // This lookup should fail, since k1 demotion would have failed
-  handle =
-      cache->Lookup(k1.AsSlice(), &LRUCacheSecondaryCacheTest::helper_fail_,
+  handle = cache->Lookup(k1.AsSlice(), GetHelperFail(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_EQ(handle, nullptr);
   // Since k1 didn't get promoted, k2 should still be in cache
-  handle =
-      cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_fail_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelperFail(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   cache->Release(handle);
@@ -1382,26 +1294,24 @@ TEST_F(LRUCacheSecondaryCacheTest, CreateFailTest) {
   Random rnd(301);
   std::string str1 = rnd.RandomString(1020);
   TestItem* item1 = new TestItem(str1.data(), str1.length());
-  ASSERT_OK(cache->Insert(k1.AsSlice(), item1,
-                          &LRUCacheSecondaryCacheTest::helper_, str1.length()));
+  ASSERT_OK(cache->Insert(k1.AsSlice(), item1, GetHelper(), str1.length()));
   std::string str2 = rnd.RandomString(1020);
   TestItem* item2 = new TestItem(str2.data(), str2.length());
   // k1 should be demoted to NVM
-  ASSERT_OK(cache->Insert(k2.AsSlice(), item2,
-                          &LRUCacheSecondaryCacheTest::helper_, str2.length()));
+  ASSERT_OK(cache->Insert(k2.AsSlice(), item2, GetHelper(), str2.length()));
 
   Cache::Handle* handle;
   SetFailCreate(true);
-  handle = cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   cache->Release(handle);
   // This lookup should fail, since k1 creation would have failed
-  handle = cache->Lookup(k1.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k1.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_EQ(handle, nullptr);
   // Since k1 didn't get promoted, k2 should still be in cache
-  handle = cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   cache->Release(handle);
@@ -1428,28 +1338,26 @@ TEST_F(LRUCacheSecondaryCacheTest, FullCapacityTest) {
   Random rnd(301);
   std::string str1 = rnd.RandomString(1020);
   TestItem* item1 = new TestItem(str1.data(), str1.length());
-  ASSERT_OK(cache->Insert(k1.AsSlice(), item1,
-                          &LRUCacheSecondaryCacheTest::helper_, str1.length()));
+  ASSERT_OK(cache->Insert(k1.AsSlice(), item1, GetHelper(), str1.length()));
   std::string str2 = rnd.RandomString(1020);
   TestItem* item2 = new TestItem(str2.data(), str2.length());
   // k1 should be demoted to NVM
-  ASSERT_OK(cache->Insert(k2.AsSlice(), item2,
-                          &LRUCacheSecondaryCacheTest::helper_, str2.length()));
+  ASSERT_OK(cache->Insert(k2.AsSlice(), item2, GetHelper(), str2.length()));
 
   Cache::Handle* handle;
-  handle = cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   // k1 promotion should fail due to the block cache being at capacity,
   // but the lookup should still succeed
   Cache::Handle* handle2;
-  handle2 = cache->Lookup(k1.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle2 = cache->Lookup(k1.AsSlice(), GetHelper(),
                           /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle2, nullptr);
   // Since k1 didn't get inserted, k2 should still be in cache
   cache->Release(handle);
   cache->Release(handle2);
-  handle = cache->Lookup(k2.AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
+  handle = cache->Lookup(k2.AsSlice(), GetHelper(),
                          /*context*/ this, Cache::Priority::LOW, true);
   ASSERT_NE(handle, nullptr);
   cache->Release(handle);
@@ -1870,8 +1778,7 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicWaitAllTest) {
     std::string str = rnd.RandomString(1020);
     values.emplace_back(str);
     TestItem* item = new TestItem(str.data(), str.length());
-    ASSERT_OK(cache->Insert(ock.WithOffset(i).AsSlice(), item,
-                            &LRUCacheSecondaryCacheTest::helper_,
+    ASSERT_OK(cache->Insert(ock.WithOffset(i).AsSlice(), item, GetHelper(),
                             str.length()));
   }
   // Force all entries to be evicted to the secondary cache
@@ -1888,9 +1795,9 @@ TEST_F(LRUCacheSecondaryCacheTest, BasicWaitAllTest) {
         TestSecondaryCache::ResultType::FAIL}});
   std::vector<Cache::Handle*> results;
   for (int i = 0; i < 6; ++i) {
-    results.emplace_back(cache->Lookup(
-        ock.WithOffset(i).AsSlice(), &LRUCacheSecondaryCacheTest::helper_,
-        /*context*/ this, Cache::Priority::LOW, false));
+    results.emplace_back(cache->Lookup(ock.WithOffset(i).AsSlice(), GetHelper(),
+                                       /*context*/ this, Cache::Priority::LOW,
+                                       false));
   }
   cache->WaitAll(results);
   for (int i = 0; i < 6; ++i) {

cache/secondary_cache.cc

@@ -32,8 +32,11 @@ Status FailCreate(const Slice&, Cache::CreateContext*, MemoryAllocator*,
 }  // namespace
 
 Status SecondaryCache::InsertSaved(const Slice& key, const Slice& saved) {
-  static Cache::CacheItemHelper helper{CacheEntryRole::kMisc, &NoopDelete,
-                                       &SliceSize, &SliceSaveTo, &FailCreate};
+  static Cache::CacheItemHelper helper_no_secondary{CacheEntryRole::kMisc,
+                                                    &NoopDelete};
+  static Cache::CacheItemHelper helper{
+      CacheEntryRole::kMisc, &NoopDelete, &SliceSize,
+      &SliceSaveTo,          &FailCreate, &helper_no_secondary};
   // NOTE: depends on Insert() being synchronous, not keeping pointer `&saved`
   return Insert(key, const_cast<Slice*>(&saved), &helper);
 }

cache/typed_cache.h

@@ -83,11 +83,14 @@ class PlaceholderCacheInterface : public BaseCacheInterface<CachePtr> {
   using BaseCacheInterface<CachePtr>::BaseCacheInterface;
 
   inline Status Insert(const Slice& key, size_t charge, Handle** handle) {
-    return this->cache_->Insert(key, /*value=*/nullptr, &kHelper, charge,
+    return this->cache_->Insert(key, /*value=*/nullptr, GetHelper(), charge,
                                 handle);
   }
 
-  static constexpr Cache::CacheItemHelper kHelper{kRole};
+  static const Cache::CacheItemHelper* GetHelper() {
+    static const Cache::CacheItemHelper kHelper{kRole};
+    return &kHelper;
+  }
 };
 
 template <CacheEntryRole kRole>
@@ -128,8 +131,11 @@ class BasicTypedCacheHelperFns {
 template <class TValue, CacheEntryRole kRole>
 class BasicTypedCacheHelper : public BasicTypedCacheHelperFns<TValue> {
  public:
-  static constexpr Cache::CacheItemHelper kBasicHelper{
-      kRole, &BasicTypedCacheHelper::Delete};
+  static const Cache::CacheItemHelper* GetBasicHelper() {
+    static const Cache::CacheItemHelper kHelper{kRole,
+                                                &BasicTypedCacheHelper::Delete};
+    return &kHelper;
+  }
 };
 
 // BasicTypedCacheInterface - Used for primary cache storage of objects of
@@ -144,7 +150,7 @@ class BasicTypedCacheInterface : public BaseCacheInterface<CachePtr>,
   CACHE_TYPE_DEFS();
   using typename BasicTypedCacheHelperFns<TValue>::TValuePtr;
   struct TypedHandle : public Handle {};
-  using BasicTypedCacheHelper<TValue, kRole>::kBasicHelper;
+  using BasicTypedCacheHelper<TValue, kRole>::GetBasicHelper;
   // ctor
   using BaseCacheInterface<CachePtr>::BaseCacheInterface;
 
@@ -154,7 +160,7 @@ class BasicTypedCacheInterface : public BaseCacheInterface<CachePtr>,
     auto untyped_handle = reinterpret_cast<Handle**>(handle);
     return this->cache_->Insert(
         key, BasicTypedCacheHelperFns<TValue>::UpCastValue(value),
-        &kBasicHelper, charge, untyped_handle, priority);
+        GetBasicHelper(), charge, untyped_handle, priority);
   }
 
   inline TypedHandle* Lookup(const Slice& key, Statistics* stats = nullptr) {
@@ -239,9 +245,16 @@ template <class TValue, class TCreateContext, CacheEntryRole kRole>
 class FullTypedCacheHelper
     : public FullTypedCacheHelperFns<TValue, TCreateContext> {
  public:
-  static constexpr Cache::CacheItemHelper kFullHelper{
-      kRole, &FullTypedCacheHelper::Delete, &FullTypedCacheHelper::Size,
-      &FullTypedCacheHelper::SaveTo, &FullTypedCacheHelper::Create};
+  static const Cache::CacheItemHelper* GetFullHelper() {
+    static const Cache::CacheItemHelper kHelper{
+        kRole,
+        &FullTypedCacheHelper::Delete,
+        &FullTypedCacheHelper::Size,
+        &FullTypedCacheHelper::SaveTo,
+        &FullTypedCacheHelper::Create,
+        BasicTypedCacheHelper<TValue, kRole>::GetBasicHelper()};
+    return &kHelper;
+  }
 };
 
 // FullTypedCacheHelper - Used for secondary cache compatible storage of
@@ -263,8 +276,8 @@ class FullTypedCacheInterface
   CACHE_TYPE_DEFS();
   using typename BasicTypedCacheInterface<TValue, kRole, CachePtr>::TypedHandle;
   using typename BasicTypedCacheHelperFns<TValue>::TValuePtr;
-  using BasicTypedCacheHelper<TValue, kRole>::kBasicHelper;
-  using FullTypedCacheHelper<TValue, TCreateContext, kRole>::kFullHelper;
+  using BasicTypedCacheHelper<TValue, kRole>::GetBasicHelper;
+  using FullTypedCacheHelper<TValue, TCreateContext, kRole>::GetFullHelper;
   using BasicTypedCacheHelperFns<TValue>::UpCastValue;
   using BasicTypedCacheHelperFns<TValue>::DownCastValue;
   // ctor
@@ -279,8 +292,8 @@ class FullTypedCacheInterface
       CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier) {
     auto untyped_handle = reinterpret_cast<Handle**>(handle);
     auto helper = lowest_used_cache_tier == CacheTier::kNonVolatileBlockTier
-                      ? &kFullHelper
-                      : &kBasicHelper;
+                      ? GetFullHelper()
+                      : GetBasicHelper();
     return this->cache_->Insert(key, UpCastValue(value), helper, charge,
                                 untyped_handle, priority);
   }
@@ -294,9 +307,9 @@ class FullTypedCacheInterface
       size_t* out_charge = nullptr) {
     ObjectPtr value;
     size_t charge;
-    Status st = kFullHelper.create_cb(data, create_context,
-                                      this->cache_->memory_allocator(), &value,
-                                      &charge);
+    Status st = GetFullHelper()->create_cb(data, create_context,
+                                           this->cache_->memory_allocator(),
+                                           &value, &charge);
     if (out_charge) {
       *out_charge = charge;
     }
@@ -304,7 +317,7 @@ class FullTypedCacheInterface
       st = InsertFull(key, DownCastValue(value), charge, nullptr /*handle*/,
                       priority, lowest_used_cache_tier);
     } else {
-      kFullHelper.del_cb(value, this->cache_->memory_allocator());
+      GetFullHelper()->del_cb(value, this->cache_->memory_allocator());
     }
     return st;
   }
@@ -318,7 +331,7 @@ class FullTypedCacheInterface
       CacheTier lowest_used_cache_tier = CacheTier::kNonVolatileBlockTier) {
     if (lowest_used_cache_tier == CacheTier::kNonVolatileBlockTier) {
       return reinterpret_cast<TypedHandle*>(this->cache_->Lookup(
-          key, &kFullHelper, create_context, priority, wait, stats));
+          key, GetFullHelper(), create_context, priority, wait, stats));
     } else {
       return BasicTypedCacheInterface<TValue, kRole, CachePtr>::Lookup(key,
                                                                        stats);

db/blob/blob_source_test.cc

@@ -1216,7 +1216,7 @@ TEST_F(BlobSecondaryCacheTest, GetBlobsFromSecondaryCache) {
       // key0's item should be in the secondary cache.
       bool kept_in_sec_cache = false;
       auto sec_handle0 = secondary_cache->Lookup(
-          key0, &BlobSource::SharedCacheInterface::kFullHelper,
+          key0, BlobSource::SharedCacheInterface::GetFullHelper(),
           /*context*/ nullptr, true,
           /*advise_erase=*/true, kept_in_sec_cache);
       ASSERT_FALSE(kept_in_sec_cache);
@@ -1244,7 +1244,7 @@ TEST_F(BlobSecondaryCacheTest, GetBlobsFromSecondaryCache) {
 
       bool kept_in_sec_cache = false;
       auto sec_handle1 = secondary_cache->Lookup(
-          key1, &BlobSource::SharedCacheInterface::kFullHelper,
+          key1, BlobSource::SharedCacheInterface::GetFullHelper(),
           /*context*/ nullptr, true,
           /*advise_erase=*/true, kept_in_sec_cache);
       ASSERT_FALSE(kept_in_sec_cache);

include/rocksdb/advanced_cache.h

@@ -134,28 +134,38 @@ class Cache {
     CreateCallback create_cb;
     // Classification of the entry for monitoring purposes in block cache.
     CacheEntryRole role;
-
-    constexpr CacheItemHelper()
-        : del_cb(nullptr),
-          size_cb(nullptr),
-          saveto_cb(nullptr),
-          create_cb(nullptr),
-          role(CacheEntryRole::kMisc) {}
-
-    explicit constexpr CacheItemHelper(CacheEntryRole _role,
-                                       DeleterFn _del_cb = nullptr,
-                                       SizeCallback _size_cb = nullptr,
-                                       SaveToCallback _saveto_cb = nullptr,
-                                       CreateCallback _create_cb = nullptr)
+    // Another CacheItemHelper (or this one) without secondary cache support.
+    // This is provided so that items promoted from secondary cache into
+    // primary cache without removal from the secondary cache can be prevented
+    // from attempting re-insertion into secondary cache (for efficiency).
+    const CacheItemHelper* without_secondary_compat;
+
+    CacheItemHelper() : CacheItemHelper(CacheEntryRole::kMisc) {}
+
+    // For helpers without SecondaryCache support
+    explicit CacheItemHelper(CacheEntryRole _role, DeleterFn _del_cb = nullptr)
+        : CacheItemHelper(_role, _del_cb, nullptr, nullptr, nullptr, this) {}
+
+    // For helpers with SecondaryCache support
+    explicit CacheItemHelper(CacheEntryRole _role, DeleterFn _del_cb,
+                             SizeCallback _size_cb, SaveToCallback _saveto_cb,
+                             CreateCallback _create_cb,
+                             const CacheItemHelper* _without_secondary_compat)
         : del_cb(_del_cb),
           size_cb(_size_cb),
           saveto_cb(_saveto_cb),
           create_cb(_create_cb),
-          role(_role) {
+          role(_role),
+          without_secondary_compat(_without_secondary_compat) {
       // Either all three secondary cache callbacks are non-nullptr or
       // all three are nullptr
       assert((size_cb != nullptr) == (saveto_cb != nullptr));
       assert((size_cb != nullptr) == (create_cb != nullptr));
+      // without_secondary_compat points to equivalent but without
+      // secondary support
+      assert(role == without_secondary_compat->role);
+      assert(del_cb == without_secondary_compat->del_cb);
+      assert(!without_secondary_compat->IsSecondaryCacheCompatible());
     }
     inline bool IsSecondaryCacheCompatible() const {
       return size_cb != nullptr;
@@ -522,6 +532,6 @@ class CacheWrapper : public Cache {
 
 // Useful for cache entries requiring no clean-up, such as for cache
 // reservations
-inline constexpr Cache::CacheItemHelper kNoopCacheItemHelper{};
+extern const Cache::CacheItemHelper kNoopCacheItemHelper;
 
 }  // namespace ROCKSDB_NAMESPACE

src.mk

@@ -382,6 +382,7 @@ TEST_LIB_SOURCES =                                              \
   db/db_test_util.cc                                            \
   db/db_with_timestamp_test_util.cc                             \
   test_util/mock_time_env.cc                                    \
+  test_util/secondary_cache_test_util.cc                        \
   test_util/testharness.cc                                      \
   test_util/testutil.cc                                         \
   utilities/agg_merge/test_agg_merge.cc                                 \

table/block_based/block_cache.cc

@@ -50,36 +50,36 @@ namespace {
 // For getting SecondaryCache-compatible helpers from a BlockType. This is
 // useful for accessing block cache in untyped contexts, such as for generic
 // cache warming in table builder.
-constexpr std::array<const Cache::CacheItemHelper*,
+const std::array<const Cache::CacheItemHelper*,
                  static_cast<unsigned>(BlockType::kInvalid) + 1>
     kCacheItemFullHelperForBlockType{{
-        &BlockCacheInterface<Block_kData>::kFullHelper,
-        &BlockCacheInterface<ParsedFullFilterBlock>::kFullHelper,
-        &BlockCacheInterface<Block_kFilterPartitionIndex>::kFullHelper,
+        BlockCacheInterface<Block_kData>::GetFullHelper(),
+        BlockCacheInterface<ParsedFullFilterBlock>::GetFullHelper(),
+        BlockCacheInterface<Block_kFilterPartitionIndex>::GetFullHelper(),
         nullptr,  // kProperties
-        &BlockCacheInterface<UncompressionDict>::kFullHelper,
-        &BlockCacheInterface<Block_kRangeDeletion>::kFullHelper,
+        BlockCacheInterface<UncompressionDict>::GetFullHelper(),
+        BlockCacheInterface<Block_kRangeDeletion>::GetFullHelper(),
         nullptr,  // kHashIndexPrefixes
         nullptr,  // kHashIndexMetadata
         nullptr,  // kMetaIndex (not yet stored in block cache)
-        &BlockCacheInterface<Block_kIndex>::kFullHelper,
+        BlockCacheInterface<Block_kIndex>::GetFullHelper(),
         nullptr,  // kInvalid
     }};
 
 // For getting basic helpers from a BlockType (no SecondaryCache support)
-constexpr std::array<const Cache::CacheItemHelper*,
+const std::array<const Cache::CacheItemHelper*,
                  static_cast<unsigned>(BlockType::kInvalid) + 1>
     kCacheItemBasicHelperForBlockType{{
-        &BlockCacheInterface<Block_kData>::kBasicHelper,
-        &BlockCacheInterface<ParsedFullFilterBlock>::kBasicHelper,
-        &BlockCacheInterface<Block_kFilterPartitionIndex>::kBasicHelper,
+        BlockCacheInterface<Block_kData>::GetBasicHelper(),
+        BlockCacheInterface<ParsedFullFilterBlock>::GetBasicHelper(),
+        BlockCacheInterface<Block_kFilterPartitionIndex>::GetBasicHelper(),
         nullptr,  // kProperties
-        &BlockCacheInterface<UncompressionDict>::kBasicHelper,
-        &BlockCacheInterface<Block_kRangeDeletion>::kBasicHelper,
+        BlockCacheInterface<UncompressionDict>::GetBasicHelper(),
+        BlockCacheInterface<Block_kRangeDeletion>::GetBasicHelper(),
         nullptr,  // kHashIndexPrefixes
         nullptr,  // kHashIndexMetadata
         nullptr,  // kMetaIndex (not yet stored in block cache)
-        &BlockCacheInterface<Block_kIndex>::kBasicHelper,
+        BlockCacheInterface<Block_kIndex>::GetBasicHelper(),
         nullptr,  // kInvalid
     }};
 }  // namespace

test_util/secondary_cache_test_util.cc

@@ -0,0 +1,94 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  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 "test_util/secondary_cache_test_util.h"
+
+#include <gtest/gtest.h>
+
+#include <array>
+
+namespace ROCKSDB_NAMESPACE {
+
+namespace secondary_cache_test_util {
+
+size_t SizeCallback(Cache::ObjectPtr obj) {
+  return static_cast<TestItem*>(obj)->Size();
+}
+
+Status SaveToCallback(Cache::ObjectPtr from_obj, size_t from_offset,
+                      size_t length, char* out) {
+  auto item = static_cast<TestItem*>(from_obj);
+  const char* buf = item->Buf();
+  EXPECT_EQ(length, item->Size());
+  EXPECT_EQ(from_offset, 0);
+  memcpy(out, buf, length);
+  return Status::OK();
+}
+
+void DeletionCallback(Cache::ObjectPtr obj, MemoryAllocator* /*alloc*/) {
+  delete static_cast<TestItem*>(obj);
+}
+
+Status SaveToCallbackFail(Cache::ObjectPtr /*obj*/, size_t /*offset*/,
+                          size_t /*size*/, char* /*out*/) {
+  return Status::NotSupported();
+}
+
+Status CreateCallback(const Slice& data, Cache::CreateContext* context,
+                      MemoryAllocator* /*allocator*/, Cache::ObjectPtr* out_obj,
+                      size_t* out_charge) {
+  auto t = static_cast<TestCreateContext*>(context);
+  if (t->fail_create_) {
+    return Status::NotSupported();
+  }
+  *out_obj = new TestItem(data.data(), data.size());
+  *out_charge = data.size();
+  return Status::OK();
+}
+
+// If helpers without_secondary are provided, returns helpers with secondary
+// support. If not provided, returns helpers without secondary support.
+static auto GenerateHelpersByRole(
+    const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>*
+        without_secondary,
+    bool fail) {
+  std::array<Cache::CacheItemHelper, kNumCacheEntryRoles> a;
+  for (uint32_t i = 0; i < kNumCacheEntryRoles; ++i) {
+    if (without_secondary) {
+      a[i] =
+          Cache::CacheItemHelper{static_cast<CacheEntryRole>(i),
+                                 &DeletionCallback,
+                                 &SizeCallback,
+                                 fail ? &SaveToCallbackFail : &SaveToCallback,
+                                 &CreateCallback,
+                                 &(*without_secondary)[i]};
+    } else {
+      a[i] = Cache::CacheItemHelper{static_cast<CacheEntryRole>(i),
+                                    &DeletionCallback};
+    }
+  }
+  return a;
+}
+
+const Cache::CacheItemHelper* GetHelper(CacheEntryRole r,
+                                        bool secondary_compatible, bool fail) {
+  static const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>
+      without_secondary = GenerateHelpersByRole(nullptr, false);
+  static const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>
+      with_secondary = GenerateHelpersByRole(&without_secondary, false);
+  static const std::array<Cache::CacheItemHelper, kNumCacheEntryRoles>
+      with_secondary_fail = GenerateHelpersByRole(&without_secondary, true);
+  return &(fail                   ? with_secondary_fail
+           : secondary_compatible ? with_secondary
+                                  : without_secondary)[static_cast<int>(r)];
+}
+
+const Cache::CacheItemHelper* GetHelperFail(CacheEntryRole r) {
+  return GetHelper(r, true, true);
+}
+
+}  // namespace secondary_cache_test_util
+
+}  // namespace ROCKSDB_NAMESPACE

test_util/secondary_cache_test_util.h

@@ -0,0 +1,56 @@
+//  Copyright (c) Meta Platforms, Inc. and affiliates.
+//  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).
+
+#pragma once
+
+#include "rocksdb/advanced_cache.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+namespace secondary_cache_test_util {
+
+class TestItem {
+ public:
+  TestItem(const char* buf, size_t size) : buf_(new char[size]), size_(size) {
+    memcpy(buf_.get(), buf, size);
+  }
+  ~TestItem() = default;
+
+  char* Buf() { return buf_.get(); }
+  [[nodiscard]] size_t Size() const { return size_; }
+  std::string ToString() { return std::string(Buf(), Size()); }
+
+ private:
+  std::unique_ptr<char[]> buf_;
+  size_t size_;
+};
+
+struct TestCreateContext : public Cache::CreateContext {
+  void SetFailCreate(bool fail) { fail_create_ = fail; }
+
+  bool fail_create_ = false;
+};
+
+size_t SizeCallback(Cache::ObjectPtr obj);
+Status SaveToCallback(Cache::ObjectPtr from_obj, size_t from_offset,
+                      size_t length, char* out);
+void DeletionCallback(Cache::ObjectPtr obj, MemoryAllocator* alloc);
+Status SaveToCallbackFail(Cache::ObjectPtr obj, size_t offset, size_t size,
+                          char* out);
+
+Status CreateCallback(const Slice& data, Cache::CreateContext* context,
+                      MemoryAllocator* allocator, Cache::ObjectPtr* out_obj,
+                      size_t* out_charge);
+
+const Cache::CacheItemHelper* GetHelper(
+    CacheEntryRole r = CacheEntryRole::kDataBlock,
+    bool secondary_compatible = true, bool fail = false);
+
+const Cache::CacheItemHelper* GetHelperFail(
+    CacheEntryRole r = CacheEntryRole::kDataBlock);
+
+}  // namespace secondary_cache_test_util
+
+}  // namespace ROCKSDB_NAMESPACE