KIKIMR-18858 Rework log cache

6 files changed, 418 insertions, 68 deletions

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp
index e8e39762067..c3e0138607a 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_blockdevice_async.cpp
@@ -1214,18 +1214,19 @@ class TCachedBlockDevice : public TRealBlockDevice {
         for (auto it = ReadsForOffset.begin(); it != ReadsForOffset.end(); it = nextIt) {
             nextIt++;
             TRead &read = it->second;
-            const TLogCache::TCacheRecord* cached = Cache.Find(read.Offset);
-            if (cached) {
-                if (read.Size <= cached->Data.Size()) {
-                    memcpy(read.Data, cached->Data.GetData(), read.Size);
-                    Mon.DeviceReadCacheHits->Inc();
-                    Y_VERIFY(read.CompletionAction);
-                    for (size_t i = 0; i < cached->BadOffsets.size(); ++i) {
-                        read.CompletionAction->RegisterBadOffset(cached->BadOffsets[i]);
-                    }
-                    NoopAsyncHackForLogReader(read.CompletionAction, read.ReqId);
-                    ReadsForOffset.erase(it);
+
+            bool foundInCache = Cache.Find(read.Offset, read.Size, static_cast<char*>(read.Data), [compAction=read.CompletionAction](auto badOffsets) {
+                for (size_t i = 0; i < badOffsets.size(); ++i) {
+                    compAction->RegisterBadOffset(badOffsets[i]);
                 }
+            });
+            
+            if (foundInCache) {
+                Mon.DeviceReadCacheHits->Inc();
+                Y_VERIFY(read.CompletionAction);
+
+                NoopAsyncHackForLogReader(read.CompletionAction, read.ReqId);
+                ReadsForOffset.erase(it);
             }
         }
         if (ReadsInFly >= MaxReadsInFly) {
@@ -1281,11 +1282,8 @@ public:
                         Cache.Pop();
                     }
                     const char* dataPtr = static_cast<const char*>(completion->GetData());
-                    Cache.Insert(
-                        TLogCache::TCacheRecord(
-                            completion->GetOffset(),
-                            TRcBuf(TString(dataPtr, dataPtr + completion->GetSize())),
-                            completion->GetBadOffsets()));
+
+                    Cache.Insert(dataPtr, completion->GetOffset(), completion->GetSize(), completion->GetBadOffsets());
                 }
             }
 
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.cpp
index 7d225c2efa9..448eb8480ae 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.cpp
@@ -27,23 +27,95 @@ size_t TLogCache::Size() const {
     return Index.size();
 }
 
-const TLogCache::TCacheRecord* TLogCache::Find(ui64 offset) {
-    auto indexIt = Index.find(offset);
-    if (indexIt == Index.end()) {
-        return nullptr;
+template <typename C>
+typename C::iterator
+FindKeyLess(C& c, const typename C::key_type& key) {
+    auto iter = c.lower_bound(key);
+    
+    if (iter == c.begin()) {
+        return c.end();
+    }
+
+    return --iter;
+}
+
+template <typename C>
+typename C::iterator
+FindKeyLessEqual(C& c, const typename C::key_type& key) {
+    auto iter = c.upper_bound(key);
+    
+    if (iter == c.begin()) {
+        return c.end();
     }
 
-    List.PushFront(&indexIt->second);
-    return &indexIt->second.Value;
+    return --iter;
 }
 
-const TLogCache::TCacheRecord* TLogCache::FindWithoutPromote(ui64 offset) const {
-    auto indexIt = Index.find(offset);
+bool TLogCache::Find(ui64 offset, ui32 size, char* buffer, TBadOffsetsHandler func) {
+    return Find(offset, size, buffer, func, true);
+}
+
+bool TLogCache::FindWithoutPromote(ui64 offset, ui32 size, char* buffer, TBadOffsetsHandler func) {
+    return Find(offset, size, buffer, func, false);
+}
+
+bool TLogCache::Find(ui64 offset, ui32 size, char* buffer, std::function<void(const std::vector<ui64>&)> func, bool promote) {
+    TVector<TItem*> res;
+
+    auto indexIt = FindKeyLessEqual(Index, offset);
+
     if (indexIt == Index.end()) {
-        return nullptr;
+        return false;
     }
-    
-    return &indexIt->second.Value;
+
+    ui64 cur = offset;
+    ui64 end = offset + size;
+
+    while (indexIt != Index.end() && cur < end) {
+        ui64 recStart = indexIt->first;
+        ui64 recEnd = recStart + indexIt->second.Value.Data.Size();
+
+        if (cur >= recStart && cur < recEnd) {
+            res.push_back(&indexIt->second);
+        } else {
+            return false;
+        }
+
+        cur = recEnd;
+
+        indexIt++;
+    }
+
+    if (cur < end) {
+        return false;
+    }
+
+    for (auto item : res) {
+        auto cacheRecord = &item->Value;
+
+        ui64 recStart = cacheRecord->Offset;
+        ui64 recEnd = recStart + cacheRecord->Data.Size();
+
+         // Determine the buffer's chunk start and end absolute offsets.
+        ui64 chunkStartOffset = std::max(recStart, offset);
+        ui64 chunkEndOffset = std::min(recEnd, offset + size);
+        ui64 chunkSize = chunkEndOffset - chunkStartOffset;
+
+        // Calculate the chunk's position within the buffer to start copying.
+        ui64 chunkOffset = chunkStartOffset - recStart;
+
+        // Copy the chunk data to the buffer.
+        std::memcpy(buffer + (chunkStartOffset - offset), cacheRecord->Data.Data() + chunkOffset, chunkSize);
+
+        // Notify callee of bad offsets.
+        func(cacheRecord->BadOffsets);
+
+        if (promote) {
+            List.PushFront(item);
+        }
+    }
+
+    return true;
 }
 
 bool TLogCache::Pop() {
@@ -55,14 +127,76 @@ bool TLogCache::Pop() {
     return true;
 }
 
-bool TLogCache::Insert(TCacheRecord&& value) {
-    auto [it, inserted] = Index.try_emplace(value.Offset, std::move(value));
-    List.PushFront(&it->second); 
-    return inserted;
+std::pair<i64, i64> TLogCache::PrepareInsertion(ui64 start, ui32 size) {
+    ui64 end = start + size;
+    ui32 leftPadding = 0;
+    ui32 rightPadding = 0;
+
+    // Check if there is a block that overlaps with the new insertion's start.
+    auto it1 = FindKeyLessEqual(Index, start);
+    if (it1 != Index.end()) {
+        ui64 maybeStart = it1->first;
+        ui64 maybeEnd = maybeStart + it1->second.Value.Data.Size();
+
+        if (start < maybeEnd) {
+            if (end <= maybeEnd) {
+                return {-1, -1}; // There is an overlapping block; return {-1, -1} to indicate it.
+            }
+            leftPadding = maybeEnd - start;
+        }
+    }
+
+    // Check if there is a block that overlaps with the new insertion's end.
+    auto it2 = FindKeyLess(Index, end);
+    if (it2 != Index.end()) {
+        ui64 maybeStart = it2->first;
+        ui64 maybeEnd = maybeStart + it2->second.Value.Data.Size();
+
+        if (end < maybeEnd) {
+            rightPadding = end - maybeStart;
+        }
+    }
+
+    // Remove any blocks that are completely covered by the new insertion.
+    ui64 offsetStart = start + leftPadding;
+    ui64 offsetEnd = start + (size - rightPadding);
+
+    auto it = Index.upper_bound(offsetStart);
+    while (it != Index.end()) {
+        ui64 blockEnd = it->first + it->second.Value.Data.Size();
+        if (blockEnd < offsetEnd) {
+            it = Index.erase(it);
+        } else {
+            break;
+        }
+    }
+
+    return {leftPadding, rightPadding};
 }
 
-size_t TLogCache::Erase(ui64 offset) {
-    return Index.erase(offset);
+bool TLogCache::Insert(const char* dataPtr, ui64 offset, ui32 size, const TVector<ui64>& badOffsets) {
+    auto [leftPadding, rightPadding] = PrepareInsertion(offset, size);
+
+    if (leftPadding == -1 && rightPadding == -1) {
+        return false;
+    }
+
+    auto dataStart = dataPtr + leftPadding;
+    auto dataEnd = dataPtr + (size - rightPadding);
+
+    Y_VERIFY_DEBUG(dataStart < dataEnd);
+
+    auto [it, inserted] = Index.try_emplace(offset + leftPadding, std::move(TLogCache::TCacheRecord(
+            offset + leftPadding,
+            TRcBuf(TString(dataStart, dataEnd)),
+            badOffsets)
+    ));
+
+    Y_VERIFY_DEBUG(inserted);
+
+    List.PushFront(&it->second); 
+
+    return true;
 }
 
 size_t TLogCache::EraseRange(ui64 begin, ui64 end) {
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.h
index c3d91dcdd2c..79bc068ff4c 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache.h
@@ -11,7 +11,7 @@ namespace NPDisk {
  * Key-value LRU cache without automatic eviction, but able to erase range of keys.
  **/
 class TLogCache {
-public:
+private:
     struct TCacheRecord {
         ui64 Offset = 0;
         TRcBuf Data;
@@ -22,7 +22,10 @@ public:
         TCacheRecord(ui64 offset, TRcBuf data, TVector<ui64> badOffsets);
     };
 
-private:
+    /**
+     * Nested class representing a cache entry in the doubly linked list.
+     * Inherits from TIntrusiveListItem to maintain the LRU order.
+     */
     struct TItem : public TIntrusiveListItem<TItem> {
         TCacheRecord Value;
 
@@ -35,19 +38,77 @@ private:
     using TIndex = TMap<ui64, TItem>;
 
 public:
+    using TBadOffsetsHandler = std::function<void(const std::vector<ui64>&)>;
+
+    /**
+     * Gets the current size of the cache.
+     */
     size_t Size() const;
-    const TCacheRecord* Find(ui64 offset);
-    const TCacheRecord* FindWithoutPromote(ui64 offset) const;
 
+    /**
+     * Finds a cache record by its offset and a specified size, copies the data to the buffer,
+     * and promotes the record to the front of the cache list.
+     * @param offset The offset key to search for.
+     * @param size The size of data to copy.
+     * @param buffer The buffer to store the copied data.
+     * @param func Optional custom function to handle bad offsets.
+     * @return True if the cache record is found and data is copied; otherwise, false.
+     */
+    bool Find(ui64 offset, ui32 size, char* buffer, TBadOffsetsHandler func = [](const std::vector<ui64>&) {});
+
+    /**
+     * Finds a cache record by its offset and a specified size, copies the data to the buffer.
+     * @param offset The offset key to search for.
+     * @param size The size of data to copy.
+     * @param buffer The buffer to store the copied data.
+     * @param func Optional custom function to handle bad offsets.
+     * @return True if the cache record is found and data is copied; otherwise, false.
+     */
+    bool FindWithoutPromote(ui64 offset, ui32 size, char* buffer, TBadOffsetsHandler func = [](const std::vector<ui64>&) {});
+
+    /**
+     * Removes the least recently used cache record from the cache.
+     * @return True if a cache record was removed; otherwise, false (cache is empty).
+     */
     bool Pop();
-    bool Insert(TCacheRecord&& value);
-    size_t Erase(ui64 offset);
-    size_t EraseRange(ui64 begin, ui64 end);  // erases range [begin, end)
+
+    /**
+     * Inserts a new cache record into the cache.
+     * @param dataPtr Pointer to the data to be inserted.
+     * @param offset The offset key for the new cache record.
+     * @param size The size of the data.
+     * @param badOffsets Optional vector of bad offsets associated with the cache record.
+     * @return True if the insertion was successful; otherwise, false (e.g., due to data being already cached).
+     */
+    bool Insert(const char* dataPtr, ui64 offset, ui32 size, const TVector<ui64>& badOffsets = {});
+
+    /**
+     * Erases a range of cache records from the cache.
+     * @param begin The beginning of the range (inclusive).
+     * @param end The end of the range (exclusive).
+     * @return The number of cache records erased.
+     */
+    size_t EraseRange(ui64 begin, ui64 end);
+
+    /**
+     * Clears the entire cache, removing all cache records.
+     */
     void Clear();
 
 private:
     TListType List;
     TIndex Index;
+
+    /**
+     * Prepares for insertion of a new cache record and calculates the left and right paddings for the data being inserted if parts of the data
+     * is already in the cache.
+     * @param offset The offset key for the new cache record.
+     * @param size The size of the data.
+     * @return A pair of i64 values representing left and right data paddings.
+     */
+    std::pair<i64, i64> PrepareInsertion(ui64 offset, ui32 size);
+
+    bool Find(ui64 offset, ui32 size, char* buffer, TBadOffsetsHandler func, bool promote);
 };
 
 } // NPDisk
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache_ut.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache_ut.cpp
index ba0d99e641a..0e268399a22 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache_ut.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_log_cache_ut.cpp
@@ -6,64 +6,216 @@ namespace NKikimr {
 namespace NPDisk {
 
 Y_UNIT_TEST_SUITE(TLogCache) {
-    TLogCache::TCacheRecord MakeRecord(ui64 offset, TString str) {
-        return TLogCache::TCacheRecord(
-            offset,
-            TRcBuf(TString(str.c_str(), str.c_str() + str.size() + 1)),
-            {});
-    }
-
     Y_UNIT_TEST(Simple) {
         TLogCache cache;
+        char buf[2] = {};
 
-        UNIT_ASSERT(cache.Insert(MakeRecord(1, "a")));
-        UNIT_ASSERT(cache.Insert(MakeRecord(2, "b")));
+        UNIT_ASSERT(cache.Insert("a", 1, 1));
+        UNIT_ASSERT(cache.Insert("b", 2, 1));
         UNIT_ASSERT_EQUAL(cache.Size(), 2);
-        UNIT_ASSERT_STRINGS_EQUAL(cache.Find(1)->Data.GetData(), "a");
+        UNIT_ASSERT(cache.Find(1, 1, buf));
+        UNIT_ASSERT_STRINGS_EQUAL(buf, "a");
 
-        UNIT_ASSERT(cache.Insert(MakeRecord(3, "c")));
+        UNIT_ASSERT(cache.Insert("c", 3, 1));
         UNIT_ASSERT(cache.Pop()); // 2 must be evicted
         UNIT_ASSERT_EQUAL(cache.Size(), 2);
-        UNIT_ASSERT_EQUAL(nullptr, cache.Find(2));
-        UNIT_ASSERT_STRINGS_EQUAL(cache.Find(3)->Data.GetData(), "c");
+        UNIT_ASSERT(!cache.Find(2, 1, buf));
+        UNIT_ASSERT(cache.Find(3, 1, buf));
+        UNIT_ASSERT_STRINGS_EQUAL(buf, "c");
 
         UNIT_ASSERT(cache.Pop()); // 1 must be evicted
-        UNIT_ASSERT(cache.Insert(MakeRecord(4, "d")));
+        UNIT_ASSERT(cache.Insert("d", 4, 1));
 
         UNIT_ASSERT_EQUAL(cache.Size(), 2);
-        UNIT_ASSERT_EQUAL(nullptr, cache.Find(1));
-        UNIT_ASSERT_STRINGS_EQUAL(cache.Find(4)->Data.GetData(), "d");
+        UNIT_ASSERT(!cache.Find(1, 1, buf));
+        UNIT_ASSERT(cache.Find(4, 1, buf));
+        UNIT_ASSERT_STRINGS_EQUAL(buf, "d");
 
         UNIT_ASSERT(cache.Pop()); // 3 must be evicted
         UNIT_ASSERT_EQUAL(cache.Size(), 1);
-        UNIT_ASSERT_EQUAL(nullptr, cache.Find(3));
-        UNIT_ASSERT_STRINGS_EQUAL(cache.Find(4)->Data.GetData(), "d");
+        UNIT_ASSERT(!cache.Find(3, 1, buf));
+        UNIT_ASSERT(cache.Find(4, 1, buf));
+        UNIT_ASSERT_STRINGS_EQUAL(buf, "d");
+
 
-        UNIT_ASSERT_EQUAL(0, cache.Erase(3));
-        UNIT_ASSERT_EQUAL(1, cache.Erase(4));
+        UNIT_ASSERT_EQUAL(1, cache.EraseRange(3, 5));
         UNIT_ASSERT_EQUAL(cache.Size(), 0);
         UNIT_ASSERT(!cache.Pop());
         UNIT_ASSERT_EQUAL(cache.Size(), 0);
     }
 
+    Y_UNIT_TEST(DoubleInsertion) {
+        TLogCache cache;
+        
+        char buf[5] = {};
+
+        auto checkFn = [&]() {
+            UNIT_ASSERT_EQUAL(25, cache.Size());
+            
+            for (int i = 0; i < 100; i += 4) {
+                UNIT_ASSERT(cache.Find(i, 4, buf));
+                UNIT_ASSERT_STRINGS_EQUAL(buf, "abcd");
+            }
+        };
+
+        for (int i = 0; i < 100; i += 4) {
+            UNIT_ASSERT(cache.Insert("abcd", i, 4));
+        }
+
+        checkFn();
+        
+        for (int i = 0; i < 100; i += 4) {
+            UNIT_ASSERT(!cache.Insert("abcd", i, 4));
+        }
+
+        checkFn();
+    }
+
+    Y_UNIT_TEST(FullyOverlapping) {
+        TLogCache cache;
+        
+        cache.Insert("abcd", 0, 4);
+        UNIT_ASSERT_EQUAL(1, cache.Size());
+
+        UNIT_ASSERT(!cache.Insert("bc", 1, 2));
+        UNIT_ASSERT_EQUAL(1, cache.Size());
+
+        char buf[2] = {};
+        UNIT_ASSERT(cache.Find(3, 1, buf));
+        UNIT_ASSERT_STRINGS_EQUAL(buf, "d");
+    }
+
+    Y_UNIT_TEST(BetweenTwoEntries) {
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("ijkl", 8, 4));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+            UNIT_ASSERT(cache.Insert("efgh", 4, 4));
+            UNIT_ASSERT_EQUAL(3, cache.Size());
+
+            char buf[5] = {};
+            UNIT_ASSERT(cache.Find(4, 4, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "efgh");
+        }
+
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("ijkl", 8, 4));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+            UNIT_ASSERT(cache.Insert("defghi", 3, 6));
+            UNIT_ASSERT_EQUAL(3, cache.Size());
+
+            char buf[5] = {};
+            UNIT_ASSERT(cache.Find(4, 4, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "efgh");
+        }
+
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("ijkl", 8, 4));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+            UNIT_ASSERT(cache.Insert("efgh", 4, 4));
+            UNIT_ASSERT_EQUAL(3, cache.Size());
+
+            char buf[7] = {};
+            UNIT_ASSERT(cache.Find(3, 6, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "defghi");
+        }
+
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("ijkl", 8, 4));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+            UNIT_ASSERT(cache.Insert("defghi", 3, 6));
+            UNIT_ASSERT_EQUAL(3, cache.Size());
+
+            char buf[7] = {};
+            UNIT_ASSERT(cache.Find(3, 6, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "defghi");
+        }
+    }
+
+    Y_UNIT_TEST(NoDuplicates) {
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("def", 3, 3));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+
+            char buf[2] = {};
+            UNIT_ASSERT(cache.Find(3, 1, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "d");
+
+            char buf2[3] = {};
+            UNIT_ASSERT(cache.Find(3, 2, buf2));
+            UNIT_ASSERT_STRINGS_EQUAL(buf2, "de");
+
+            char buf3[11] = {};
+            UNIT_ASSERT(!cache.Find(3, 10, buf3));
+            UNIT_ASSERT_STRINGS_EQUAL(buf3, "");
+        }
+
+        {
+            TLogCache cache;
+            
+            UNIT_ASSERT(cache.Insert("def", 3, 3));
+            UNIT_ASSERT_EQUAL(1, cache.Size());
+            UNIT_ASSERT(cache.Insert("abcd", 0, 4));
+            UNIT_ASSERT_EQUAL(2, cache.Size());
+
+            char buf[2] = {};
+            UNIT_ASSERT(cache.Find(3, 1, buf));
+            UNIT_ASSERT_STRINGS_EQUAL(buf, "d");
+
+            char buf2[5] = {};
+            UNIT_ASSERT(cache.Find(0, 4, buf2));
+            UNIT_ASSERT_STRINGS_EQUAL(buf2, "abcd");
+
+            char buf3[11] = {};
+            UNIT_ASSERT(!cache.Find(3, 10, buf3));
+            UNIT_ASSERT_STRINGS_EQUAL(buf3, "");
+        }
+    }
+
     TLogCache SetupCache(const TVector<std::pair<ui64, TString>>& content = {{5, "x"}, {1, "y"}, {10, "z"}}) {
         TLogCache cache;
         for (auto pair : content) {
-            cache.Insert(MakeRecord(pair.first, pair.second));
+            auto& data = pair.second;
+
+            cache.Insert(data.c_str(), pair.first, data.Size());
         }
         return cache;
     };
 
     void AssertCacheContains(TLogCache& cache, const TVector<std::pair<ui64, TString>>& content = {{5, "x"}, {1, "y"}, {10, "z"}}) {
         UNIT_ASSERT_VALUES_EQUAL(content.size(), cache.Size());
+
+        char buf[2] = {};
+
         for (auto pair : content) {
-            UNIT_ASSERT_STRINGS_EQUAL(
-                pair.second,
-                cache.FindWithoutPromote(pair.first)->Data.GetData());
+            UNIT_ASSERT(cache.FindWithoutPromote(pair.first, 1, buf));
+
+            UNIT_ASSERT_STRINGS_EQUAL(pair.second, buf);
         }
+
         for (auto pair : content) {
             UNIT_ASSERT(cache.Pop());
-            UNIT_ASSERT_EQUAL(nullptr, cache.FindWithoutPromote(pair.first));
+
+            UNIT_ASSERT(!cache.FindWithoutPromote(pair.first, 1, buf));
         }
     }
 
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_helpers.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_helpers.cpp
index 4a79eaedfe3..7c574e26d92 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_helpers.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_helpers.cpp
@@ -12,11 +12,16 @@
 namespace NKikimr {
 
 TString PrepareData(ui32 size, ui32 flavor) {
-    TString data = TString::Uninitialized(size);
+    TString str = TString::Uninitialized(size);
+    
+    // Using char* enables possibility to vectorize the following loop.
+    char* data = str.Detach();
+
     for (ui32 i = 0; i < size; ++i) {
         data[i] = '0' + (i + size + flavor) % 8;
     }
-    return data;
+
+    return str;
 }
 
 TString StatusToString(const NKikimrProto::EReplyStatus status) {
diff --git a/ydb/library/pdisk_io/wcache.cpp b/ydb/library/pdisk_io/wcache.cpp
index 0fc52e79af6..1877a3d09f5 100644
--- a/ydb/library/pdisk_io/wcache.cpp
+++ b/ydb/library/pdisk_io/wcache.cpp
@@ -412,7 +412,7 @@ struct TIdentifyData {
     static const ui32 IdentifySizeBytes = 512;
     bool IsGathered = false;
     ui8 Data[IdentifySizeBytes];
-    // Offset in words, descirption, size bytes for strings
+    // Offset in words, description, size bytes for strings
     // 10, serial number, 20 ASCII
     // 23, firmware revision, 8 ASCII
     // 27, Model number, 40 ASCII