Fix a potential memory leak on row_cache insertion failure (#11682)

Summary:
Although the built-in Cache implementations never return failure on Insert without keeping a reference (Handle), a custom implementation could. The code for inserting into row_cache does not keep a reference but does not clean up appropriately on non-OK. This is a fix.

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

Test Plan: unit test added that previously fails under ASAN

Reviewed By: ajkr

Differential Revision: D48153831

Pulled By: pdillinger

fbshipit-source-id: 86eb7387915c5b38b6ff5dd8deb4e1e223b7d020

db/db_test.cc

@@ -6927,6 +6927,27 @@ TEST_F(DBTest, RowCache) {
   ASSERT_EQ(Get("foo"), "bar");
   ASSERT_EQ(TestGetTickerCount(options, ROW_CACHE_HIT), 1);
   ASSERT_EQ(TestGetTickerCount(options, ROW_CACHE_MISS), 1);
+
+  // Also test non-OK cache insertion (would be ASAN failure on memory leak)
+  class FailInsertionCache : public CacheWrapper {
+   public:
+    using CacheWrapper::CacheWrapper;
+    const char* Name() const override { return "FailInsertionCache"; }
+    Status Insert(const Slice&, Cache::ObjectPtr, const CacheItemHelper*,
+                  size_t, Handle** = nullptr,
+                  Priority = Priority::LOW) override {
+      return Status::MemoryLimit();
+    }
+  };
+  options.row_cache = std::make_shared<FailInsertionCache>(options.row_cache);
+  ASSERT_OK(options.statistics->Reset());
+  Reopen(options);
+
+  ASSERT_EQ(Get("foo"), "bar");
+  ASSERT_EQ(TestGetTickerCount(options, ROW_CACHE_MISS), 1);
+  ASSERT_EQ(Get("foo"), "bar");
+  // Test condition requires row cache insertion to fail
+  ASSERT_EQ(TestGetTickerCount(options, ROW_CACHE_MISS), 2);
 }
 
 TEST_F(DBTest, PinnableSliceAndRowCache) {

db/table_cache.cc

@@ -484,9 +484,12 @@ Status TableCache::Get(
     RowCacheInterface row_cache{ioptions_.row_cache.get()};
     size_t charge = row_cache_entry->capacity() + sizeof(std::string);
     auto row_ptr = new std::string(std::move(*row_cache_entry));
-    // If row cache is full, it's OK to continue.
-    row_cache.Insert(row_cache_key.GetUserKey(), row_ptr, charge)
-        .PermitUncheckedError();
+    Status rcs = row_cache.Insert(row_cache_key.GetUserKey(), row_ptr, charge);
+    if (!rcs.ok()) {
+      // If row cache is full, it's OK to continue, but we keep ownership of
+      // row_ptr.
+      delete row_ptr;
+    }
   }
 
   if (handle != nullptr) {