YQL-19767: Introduce MKQL allocator address sanitizing

Здесь поддержал только out of bounds access и use after free. Отложенное использование памяти и тд буду делать потом
commit_hash:2a3fd472b626762ff7c8b7b0bc1285af50c511cf

19 files changed, 633 insertions, 224 deletions

diff --git a/yql/essentials/minikql/aligned_page_pool.cpp b/yql/essentials/minikql/aligned_page_pool.cpp
index e061c08574d..f98065f20e5 100644
--- a/yql/essentials/minikql/aligned_page_pool.cpp
+++ b/yql/essentials/minikql/aligned_page_pool.cpp
@@ -10,6 +10,8 @@
 #include <util/system/info.h>
 #include <util/thread/lfstack.h>
 
+#include <yql/essentials/minikql/asan_utils.h>
+
 #if defined(_win_)
 #   include <util/system/winint.h>
 #elif defined(_unix_)
@@ -76,6 +78,7 @@ public:
         void *page = nullptr;
         if (Pages.Dequeue(&page)) {
             --Count;
+            SanitizerMarkInvalid(page, PageSize);
             return page;
         }
 
@@ -101,13 +104,14 @@ private:
             FreePage(addr);
             return GetPageSize();
         }
-
+        SanitizerMarkInvalid(addr, PageSize);
         ++Count;
         Pages.Enqueue(addr);
         return 0;
     }
 
     void FreePage(void* addr) {
+        SanitizerMarkInvalid(addr, PageSize);
         auto res = T::Munmap(addr, PageSize);
         Y_DEBUG_ABORT_UNLESS(0 == res, "Madvise failed: %s", LastSystemErrorText());
     }
@@ -142,6 +146,7 @@ public:
         Y_DEBUG_ABORT_UNLESS(!TAlignedPagePoolImpl<T>::IsDefaultAllocatorUsed(), "No memory maps allowed while using default allocator");
 
         void* res = T::Mmap(size);
+        SanitizerMarkInvalid(res, size);
         TotalMmappedBytes += size;
         return res;
     }
@@ -289,38 +294,15 @@ inline int TSystemMmap::Munmap(void* addr, size_t size)
 }
 #endif
 
-std::function<void(size_t size)> TFakeAlignedMmap::OnMmap = {};
-std::function<void(void* addr, size_t size)> TFakeAlignedMmap::OnMunmap = {};
-
-void* TFakeAlignedMmap::Mmap(size_t size)
-{
-    if (OnMmap) {
-        OnMmap(size);
-    }
-    return reinterpret_cast<void*>(TAlignedPagePool::POOL_PAGE_SIZE);
-}
-
-int TFakeAlignedMmap::Munmap(void* addr, size_t size)
-{
-    if (OnMunmap) {
-        OnMunmap(addr, size);
-    }
-    return 0;
-}
-
-std::function<void(size_t size)> TFakeUnalignedMmap::OnMmap = {};
-std::function<void(void* addr, size_t size)> TFakeUnalignedMmap::OnMunmap = {};
+std::function<void*(size_t size)> TFakeMmap::OnMmap = {};
+std::function<void(void* addr, size_t size)> TFakeMmap::OnMunmap = {};
 
-void* TFakeUnalignedMmap::Mmap(size_t size)
-{
-    if (OnMmap) {
-        OnMmap(size);
-    }
-    return reinterpret_cast<void*>(TAlignedPagePool::POOL_PAGE_SIZE+1);
+void* TFakeMmap::Mmap(size_t size) {
+    Y_DEBUG_ABORT_UNLESS(OnMmap, "mmap function must be provided");
+    return OnMmap(size);
 }
 
-int TFakeUnalignedMmap::Munmap(void* addr, size_t size)
-{
+int TFakeMmap::Munmap(void* addr, size_t size) {
     if (OnMunmap) {
         OnMunmap(addr, size);
     }
@@ -436,7 +418,7 @@ void TAlignedPagePoolImpl<T>::OffloadFree(ui64 size) noexcept {
 }
 
 template<typename T>
-void* TAlignedPagePoolImpl<T>::GetPage() {
+void* TAlignedPagePoolImpl<T>::GetPageImpl() {
     ++PageAllocCount;
     if (!FreePages.empty()) {
         ++PageHitCount;
@@ -479,6 +461,13 @@ void* TAlignedPagePoolImpl<T>::GetPage() {
     return res;
 }
 
+template <typename T>
+void* TAlignedPagePoolImpl<T>::GetPage() {
+    auto* page = GetPageImpl();
+    SanitizerMarkInvalid(page, POOL_PAGE_SIZE);
+    return page;
+};
+
 template<typename T>
 void TAlignedPagePoolImpl<T>::ReturnPage(void* addr) noexcept {
     if (Y_UNLIKELY(IsDefaultAllocatorUsed())) {
@@ -487,6 +476,7 @@ void TAlignedPagePoolImpl<T>::ReturnPage(void* addr) noexcept {
         return;
     }
 
+    SanitizerMarkInvalid(addr, POOL_PAGE_SIZE);
     Y_DEBUG_ABORT_UNLESS(AllPages.find(addr) != AllPages.end());
     FreePages.emplace(addr);
 }
@@ -504,14 +494,9 @@ void* TAlignedPagePoolImpl<T>::GetBlock(size_t size) {
 
         return ret;
     }
-
-    if (size == POOL_PAGE_SIZE) {
-        return GetPage();
-    } else {
-        const auto ptr = Alloc(size);
-        Y_DEBUG_ABORT_UNLESS(ActiveBlocks.emplace(ptr, size).second);
-        return ptr;
-    }
+    auto* block = GetBlockImpl(size);
+    SanitizerMarkInvalid(block, size);
+    return block;
 }
 
 template<typename T>
@@ -681,6 +666,17 @@ bool TAlignedPagePoolImpl<T>::TryIncreaseLimit(ui64 required) {
     return Limit >= required;
 }
 
+template <typename T>
+void* TAlignedPagePoolImpl<T>::GetBlockImpl(size_t size) {
+    if (size == POOL_PAGE_SIZE) {
+        return GetPage();
+    } else {
+        const auto ptr = Alloc(size);
+        Y_DEBUG_ABORT_UNLESS(ActiveBlocks.emplace(ptr, size).second);
+        return ptr;
+    }
+}
+
 template<typename T>
 ui64 TAlignedPagePoolImpl<T>::GetGlobalPagePoolSize() {
     ui64 size = 0;
@@ -713,8 +709,7 @@ bool TAlignedPagePoolImpl<T>::IsDefaultAllocatorUsed() {
 #endif
 
 template class TAlignedPagePoolImpl<>;
-template class TAlignedPagePoolImpl<TFakeAlignedMmap>;
-template class TAlignedPagePoolImpl<TFakeUnalignedMmap>;
+template class TAlignedPagePoolImpl<TFakeMmap>;
 
 template<typename TMmap>
 void* GetAlignedPage(ui64 size) {
@@ -809,7 +804,6 @@ void ReleaseAlignedPage(void* mem, ui64 size) {
         TGlobalPools<TMmap, true>::Instance().PushPage(level, mem);
         return;
     }
-
     TGlobalPools<TMmap, true>::Instance().DoMunmap(mem, size);
 }
 
@@ -829,28 +823,22 @@ i64 GetTotalFreeListBytes() {
 }
 
 template i64 GetTotalMmapedBytes<>();
-template i64 GetTotalMmapedBytes<TFakeAlignedMmap>();
-template i64 GetTotalMmapedBytes<TFakeUnalignedMmap>();
+template i64 GetTotalMmapedBytes<TFakeMmap>();
 
 template i64 GetTotalFreeListBytes<>();
-template i64 GetTotalFreeListBytes<TFakeAlignedMmap>();
-template i64 GetTotalFreeListBytes<TFakeUnalignedMmap>();
+template i64 GetTotalFreeListBytes<TFakeMmap>();
 
 template void* GetAlignedPage<>(ui64);
-template void* GetAlignedPage<TFakeAlignedMmap>(ui64);
-template void* GetAlignedPage<TFakeUnalignedMmap>(ui64);
+template void* GetAlignedPage<TFakeMmap>(ui64);
 
 template void* GetAlignedPage<>();
-template void* GetAlignedPage<TFakeAlignedMmap>();
-template void* GetAlignedPage<TFakeUnalignedMmap>();
+template void* GetAlignedPage<TFakeMmap>();
 
 template void ReleaseAlignedPage<>(void*,ui64);
-template void ReleaseAlignedPage<TFakeAlignedMmap>(void*,ui64);
-template void ReleaseAlignedPage<TFakeUnalignedMmap>(void*,ui64);
+template void ReleaseAlignedPage<TFakeMmap>(void*,ui64);
 
 template void ReleaseAlignedPage<>(void*);
-template void ReleaseAlignedPage<TFakeAlignedMmap>(void*);
-template void ReleaseAlignedPage<TFakeUnalignedMmap>(void*);
+template void ReleaseAlignedPage<TFakeMmap>(void*);
 
 size_t GetMemoryMapsCount() {
     size_t lineCount = 0;
diff --git a/yql/essentials/minikql/aligned_page_pool.h b/yql/essentials/minikql/aligned_page_pool.h
index bc570528c97..6028878f5a9 100644
--- a/yql/essentials/minikql/aligned_page_pool.h
+++ b/yql/essentials/minikql/aligned_page_pool.h
@@ -9,6 +9,8 @@
 #include <util/system/yassert.h>
 
 #include <stack>
+#include <queue>
+
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
@@ -50,18 +52,9 @@ public:
     static int Munmap(void* addr, size_t size);
 };
 
-class TFakeAlignedMmap {
+class TFakeMmap {
 public:
-    static std::function<void(size_t size)> OnMmap;
-    static std::function<void(void* addr, size_t size)> OnMunmap;
-
-    static void* Mmap(size_t size);
-    static int Munmap(void* addr, size_t size);
-};
-
-class TFakeUnalignedMmap {
-public:
-    static std::function<void(size_t size)> OnMmap;
+    static std::function<void*(size_t size)> OnMmap;
     static std::function<void(void* addr, size_t size)> OnMunmap;
 
     static void* Mmap(size_t size);
@@ -261,6 +254,9 @@ protected:
 
     bool TryIncreaseLimit(ui64 required);
 
+    void* GetBlockImpl(size_t size);
+
+    void* GetPageImpl();
 protected:
     std::stack<void*, std::vector<void*>> FreePages;
     std::unordered_set<void*> AllPages;
diff --git a/yql/essentials/minikql/aligned_page_pool_ut.cpp b/yql/essentials/minikql/aligned_page_pool_ut.cpp
index b917176dd12..d586d002e9e 100644
--- a/yql/essentials/minikql/aligned_page_pool_ut.cpp
+++ b/yql/essentials/minikql/aligned_page_pool_ut.cpp
@@ -3,139 +3,156 @@
 #include <library/cpp/testing/unittest/registar.h>
 
 #include <util/system/info.h>
+#include <yql/essentials/utils/backtrace/backtrace.h>
 
 namespace NKikimr {
 namespace NMiniKQL {
 
-Y_UNIT_TEST_SUITE(TAlignedPagePoolTest) {
-
-Y_UNIT_TEST(AlignedMmapPageSize) {
-    TAlignedPagePool::ResetGlobalsUT();
-    TAlignedPagePoolImpl<TFakeAlignedMmap> alloc(__LOCATION__);
+namespace {
+class TScopedMemoryMapper {
+public:
+    static constexpr size_t EXTRA_SPACE_FOR_UNALIGNMENT = 1;
 
-    int munmaps = 0;
-    TFakeAlignedMmap::OnMunmap = [&](void* addr, size_t s) {
-        Y_UNUSED(addr);
-        Y_UNUSED(s);
-        munmaps ++;
+    struct TUnmapEntry {
+        void* Addr;
+        size_t Size;
+        bool operator==(const TUnmapEntry& rhs) {
+            return std::tie(Addr, Size) == std::tie(rhs.Addr, rhs.Size);
+        }
     };
 
+    TScopedMemoryMapper(bool aligned) {
+        Aligned_ = aligned;
+        TFakeMmap::OnMunmap = [this](void* addr, size_t s) {
+            Munmaps_.push_back({addr, s});
+        };
+
+        TFakeMmap::OnMmap = [this](size_t size) -> void* {
+            // Allocate more memory to ensure we have enough space for alignment
+            Storage_ = THolder<char, TDeleteArray>(new char[AlignUp(size + EXTRA_SPACE_FOR_UNALIGNMENT, TAlignedPagePool::POOL_PAGE_SIZE)]);
+            UNIT_ASSERT(Storage_.Get());
+
+            // Force TFakeMmap::Munmap to be called by returning a pointer that will always need adjustment
+            if (Aligned_) {
+                return PointerToAlignedMemory();
+            } else {
+                // Ensure the pointer is always unaligned by a fixed amount
+                void* ptr = PointerToAlignedMemory();
+                // Add EXTRA_SPACE_FOR_UNALIGNMENT to ensure it's unaligned
+                return static_cast<void*>(static_cast<char*>(ptr) + EXTRA_SPACE_FOR_UNALIGNMENT);
+            }
+        };
+    }
+
+    ~TScopedMemoryMapper() {
+        TFakeMmap::OnMunmap = {};
+        TFakeMmap::OnMmap = {};
+        Storage_.Reset();
+    }
+
+    void* PointerToAlignedMemory() {
+        return AlignUp(Storage_.Get(), TAlignedPagePool::POOL_PAGE_SIZE);
+    }
+
+    size_t MunmapsSize() {
+        return Munmaps_.size();
+    }
+
+    TUnmapEntry Munmaps(size_t i) {
+        return Munmaps_[i];
+    }
+
+private:
+    THolder<char, TDeleteArray> Storage_;
+    std::vector<TUnmapEntry> Munmaps_;
+    bool Aligned_;
+};
+
+}; // namespace
+
+Y_UNIT_TEST_SUITE(TAlignedPagePoolTest) {
+
+Y_UNIT_TEST(AlignedMmapPageSize) {
+    TAlignedPagePoolImpl<TFakeMmap>::ResetGlobalsUT();
+    TAlignedPagePoolImpl<TFakeMmap> alloc(__LOCATION__);
+    TScopedMemoryMapper mmapper(/*aligned=*/true);
     auto size = TAlignedPagePool::POOL_PAGE_SIZE;
     auto block = std::shared_ptr<void>(alloc.GetBlock(size), [&](void* addr) { alloc.ReturnBlock(addr, size); });
-    TFakeAlignedMmap::OnMunmap = {};
-    UNIT_ASSERT_EQUAL(0, munmaps);
+    UNIT_ASSERT_EQUAL(0u, mmapper.MunmapsSize());
 
-    UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(block.get()), TAlignedPagePool::POOL_PAGE_SIZE);
+    UNIT_ASSERT_VALUES_EQUAL(block.get(), mmapper.PointerToAlignedMemory());
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount()
-        , TAlignedPagePool::ALLOC_AHEAD_PAGES);
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount(), TAlignedPagePool::ALLOC_AHEAD_PAGES);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated()
-        , TAlignedPagePool::POOL_PAGE_SIZE + TAlignedPagePool::ALLOC_AHEAD_PAGES*TAlignedPagePool::POOL_PAGE_SIZE
-    );
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated(), TAlignedPagePool::POOL_PAGE_SIZE + TAlignedPagePool::ALLOC_AHEAD_PAGES * TAlignedPagePool::POOL_PAGE_SIZE);
 }
 
 Y_UNIT_TEST(UnalignedMmapPageSize) {
-    TAlignedPagePool::ResetGlobalsUT();
-    TAlignedPagePoolImpl<TFakeUnalignedMmap> alloc(__LOCATION__);
-
-    int munmaps = 0;
-    TFakeUnalignedMmap::OnMunmap = [&](void* addr, size_t s) {
-        Y_UNUSED(addr);
-        if (munmaps == 0) {
-            UNIT_ASSERT_VALUES_EQUAL(s, TAlignedPagePool::POOL_PAGE_SIZE - 1);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(s, 1);
-        }
-        munmaps ++;
-    };
+    TAlignedPagePoolImpl<TFakeMmap>::ResetGlobalsUT();
+    TAlignedPagePoolImpl<TFakeMmap> alloc(__LOCATION__);
+    TScopedMemoryMapper mmapper(/*aligned=*/false);
 
     auto size = TAlignedPagePool::POOL_PAGE_SIZE;
     auto block = std::shared_ptr<void>(alloc.GetBlock(size), [&](void* addr) { alloc.ReturnBlock(addr, size); });
-    TFakeUnalignedMmap::OnMunmap = {};
-    UNIT_ASSERT_EQUAL(2, munmaps);
+    UNIT_ASSERT_EQUAL(2, mmapper.MunmapsSize());
+    UNIT_ASSERT_EQUAL(TAlignedPagePool::POOL_PAGE_SIZE - TScopedMemoryMapper::EXTRA_SPACE_FOR_UNALIGNMENT, mmapper.Munmaps(0).Size);
+    UNIT_ASSERT_EQUAL(TScopedMemoryMapper::EXTRA_SPACE_FOR_UNALIGNMENT, mmapper.Munmaps(1).Size);
 
-    UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(block.get()), 2 * TAlignedPagePool::POOL_PAGE_SIZE);
+    UNIT_ASSERT_VALUES_EQUAL(block.get(), (char*)mmapper.PointerToAlignedMemory() + TAlignedPagePool::POOL_PAGE_SIZE);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount()
-        , TAlignedPagePool::ALLOC_AHEAD_PAGES - 1);
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount(), TAlignedPagePool::ALLOC_AHEAD_PAGES - 1);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated()
-        , TAlignedPagePool::POOL_PAGE_SIZE + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 1) * TAlignedPagePool::POOL_PAGE_SIZE
-    );
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated(), TAlignedPagePool::POOL_PAGE_SIZE + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 1) * TAlignedPagePool::POOL_PAGE_SIZE);
 }
 
 Y_UNIT_TEST(AlignedMmapUnalignedSize) {
-    TAlignedPagePool::ResetGlobalsUT();
-    TAlignedPagePoolImpl<TFakeAlignedMmap> alloc(__LOCATION__);
+    TAlignedPagePoolImpl<TFakeMmap>::ResetGlobalsUT();
+    TAlignedPagePoolImpl<TFakeMmap> alloc(__LOCATION__);
     auto smallSize = NSystemInfo::GetPageSize();
     auto size = smallSize + 1024 * TAlignedPagePool::POOL_PAGE_SIZE;
-
-    int munmaps = 0;
-    TFakeAlignedMmap::OnMunmap = [&](void* addr, size_t s) {
-        if (munmaps == 0) {
-            UNIT_ASSERT_VALUES_EQUAL(s, TAlignedPagePool::POOL_PAGE_SIZE - smallSize);
-            UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(addr), TAlignedPagePool::POOL_PAGE_SIZE + size);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(s, smallSize);
-            UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(addr), TAlignedPagePool::POOL_PAGE_SIZE + TAlignedPagePool::ALLOC_AHEAD_PAGES * TAlignedPagePool::POOL_PAGE_SIZE + size - smallSize);
-        }
-
-        munmaps ++;
-    };
+    TScopedMemoryMapper mmapper(/*aligned=*/true);
 
     auto block = std::shared_ptr<void>(alloc.GetBlock(size), [&](void* addr) { alloc.ReturnBlock(addr, size); });
-    TFakeAlignedMmap::OnMunmap = {};
 
-    UNIT_ASSERT_EQUAL(2, munmaps);
+    UNIT_ASSERT_EQUAL(2, mmapper.MunmapsSize());
+    auto expected0 = (TScopedMemoryMapper::TUnmapEntry{(char*)mmapper.PointerToAlignedMemory() + size, TAlignedPagePool::POOL_PAGE_SIZE - smallSize});
+    UNIT_ASSERT_EQUAL(expected0, mmapper.Munmaps(0));
+    auto expected1 = TScopedMemoryMapper::TUnmapEntry{
+        (char*)mmapper.PointerToAlignedMemory() + TAlignedPagePool::ALLOC_AHEAD_PAGES * TAlignedPagePool::POOL_PAGE_SIZE + size - smallSize,
+        smallSize};
+    UNIT_ASSERT_EQUAL(expected1, mmapper.Munmaps(1));
 
-    UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(block.get()), TAlignedPagePool::POOL_PAGE_SIZE);
+    UNIT_ASSERT_VALUES_EQUAL(block.get(), mmapper.PointerToAlignedMemory());
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount()
-        , TAlignedPagePool::ALLOC_AHEAD_PAGES - 1);
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount(), TAlignedPagePool::ALLOC_AHEAD_PAGES - 1);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated()
-        , size + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 1) * TAlignedPagePool::POOL_PAGE_SIZE
-    );
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated(), size + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 1) * TAlignedPagePool::POOL_PAGE_SIZE);
 }
 
 Y_UNIT_TEST(UnalignedMmapUnalignedSize) {
-    TAlignedPagePool::ResetGlobalsUT();
-    TAlignedPagePoolImpl<TFakeUnalignedMmap> alloc(__LOCATION__);
+    TAlignedPagePoolImpl<TFakeMmap>::ResetGlobalsUT();
+    TAlignedPagePoolImpl<TFakeMmap> alloc(__LOCATION__);
     auto smallSize = NSystemInfo::GetPageSize();
     auto size = smallSize + 1024 * TAlignedPagePool::POOL_PAGE_SIZE;
-    int munmaps = 0;
-    TFakeUnalignedMmap::OnMunmap = [&](void* addr, size_t s) {
-        Y_UNUSED(addr);
-        if (munmaps == 0) {
-            UNIT_ASSERT_VALUES_EQUAL(s, TAlignedPagePool::POOL_PAGE_SIZE - 1);
-        } else if (munmaps == 1) {
-            UNIT_ASSERT_VALUES_EQUAL(s, TAlignedPagePool::POOL_PAGE_SIZE - smallSize);
-        } else {
-            UNIT_ASSERT_VALUES_EQUAL(s, smallSize + 1);
-        }
-        munmaps ++;
-    };
-
+    TScopedMemoryMapper mmapper(/*aligned=*/false);
     auto block = std::shared_ptr<void>(alloc.GetBlock(size), [&](void* addr) { alloc.ReturnBlock(addr, size); });
-    TFakeUnalignedMmap::OnMunmap = {};
-    UNIT_ASSERT_EQUAL(3, munmaps);
+    UNIT_ASSERT_EQUAL(3, mmapper.MunmapsSize());
+    UNIT_ASSERT_EQUAL(TAlignedPagePool::POOL_PAGE_SIZE - TScopedMemoryMapper::EXTRA_SPACE_FOR_UNALIGNMENT, mmapper.Munmaps(0).Size);
+    UNIT_ASSERT_EQUAL(TAlignedPagePool::POOL_PAGE_SIZE - smallSize, mmapper.Munmaps(1).Size);
+    UNIT_ASSERT_EQUAL(smallSize + 1, mmapper.Munmaps(2).Size);
 
-    UNIT_ASSERT_VALUES_EQUAL(reinterpret_cast<uintptr_t>(block.get()), 2 * TAlignedPagePool::POOL_PAGE_SIZE);
+    UNIT_ASSERT_VALUES_EQUAL(block.get(), (char*)mmapper.PointerToAlignedMemory() + TAlignedPagePool::POOL_PAGE_SIZE);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount()
-        , TAlignedPagePool::ALLOC_AHEAD_PAGES - 2);
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreePageCount(), TAlignedPagePool::ALLOC_AHEAD_PAGES - 2);
 
-    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated()
-        , size + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 2) * TAlignedPagePool::POOL_PAGE_SIZE
-    );
+    UNIT_ASSERT_VALUES_EQUAL(alloc.GetAllocated(), size + (TAlignedPagePool::ALLOC_AHEAD_PAGES - 2) * TAlignedPagePool::POOL_PAGE_SIZE);
 }
 
 Y_UNIT_TEST(YellowZoneSwitchesCorrectlyBlock) {
     TAlignedPagePool::ResetGlobalsUT();
     TAlignedPagePoolImpl alloc(__LOCATION__);
 
-    // choose relatively big chunk so ALLOC_AHEAD_PAGES don't affect the correctness of the test 
+    // choose relatively big chunk so ALLOC_AHEAD_PAGES don't affect the correctness of the test
     auto size = 1024 * TAlignedPagePool::POOL_PAGE_SIZE;
 
     alloc.SetLimit(size * 10);
diff --git a/yql/essentials/minikql/asan_utils.h b/yql/essentials/minikql/asan_utils.h
new file mode 100644
index 00000000000..b404be7b635
--- /dev/null
+++ b/yql/essentials/minikql/asan_utils.h
@@ -0,0 +1,91 @@
+#pragma once
+
+#include <cstddef>
+
+#include <util/system/compiler.h>
+#include <util/system/yassert.h>
+
+#if defined(_asan_enabled_)
+    #include <sanitizer/asan_interface.h>
+#endif
+
+namespace NKikimr {
+
+inline constexpr size_t ALLOCATION_REDZONE_SIZE = 16;
+inline constexpr size_t ASAN_EXTRA_ALLOCATION_SPACE = ALLOCATION_REDZONE_SIZE * 2;
+
+constexpr void* SanitizerMarkInvalid(void* addr, size_t size) {
+#if defined(_asan_enabled_)
+    if (addr == nullptr) {
+        return nullptr;
+    }
+    __asan_poison_memory_region(addr, size);
+#else // defined(_asan_enabled_)
+    Y_UNUSED(addr, size);
+#endif
+    return addr;
+}
+
+constexpr void* SanitizerMarkValid(void* addr, size_t size) {
+#if defined(_asan_enabled_)
+    if (addr == nullptr) {
+        return nullptr;
+    }
+    __asan_unpoison_memory_region(addr, size);
+#else // defined(_asan_enabled_)
+    Y_UNUSED(addr, size);
+#endif
+    return addr;
+}
+
+constexpr size_t GetSizeToAlloc(size_t size) {
+#if defined(_asan_enabled_)
+    if (size == 0) {
+        return 0;
+    }
+    return size + 2 * ALLOCATION_REDZONE_SIZE;
+#else // defined(_asan_enabled_)
+    return size;
+#endif
+}
+
+constexpr const void* GetOriginalAllocatedObject(const void* ptr, size_t size) {
+#if defined(_asan_enabled_)
+    if (size == 0) {
+        return ptr;
+    }
+    return (char*)ptr - ALLOCATION_REDZONE_SIZE;
+#else // defined(_asan_enabled_)
+    Y_UNUSED(size);
+    return ptr;
+#endif
+}
+
+constexpr void* WrapPointerWithRedZones(void* ptr, size_t extendedSizeWithRedzone) {
+#if defined(_asan_enabled_)
+    if (extendedSizeWithRedzone == 0) {
+        return ptr;
+    }
+    SanitizerMarkInvalid(ptr, extendedSizeWithRedzone);
+    SanitizerMarkValid((char*)ptr + ALLOCATION_REDZONE_SIZE, extendedSizeWithRedzone - 2 * ALLOCATION_REDZONE_SIZE);
+    return (char*)ptr + ALLOCATION_REDZONE_SIZE;
+#else // defined(_asan_enabled_)
+    Y_UNUSED(extendedSizeWithRedzone);
+    return ptr;
+#endif
+}
+
+constexpr const void* UnwrapPointerWithRedZones(const void* ptr, size_t size) {
+#if defined(_asan_enabled_)
+    if (size == 0) {
+        return ptr;
+    }
+    SanitizerMarkInvalid((char*)ptr - ALLOCATION_REDZONE_SIZE, 2 * ALLOCATION_REDZONE_SIZE + size);
+    return (char*)ptr - ALLOCATION_REDZONE_SIZE;
+#else // defined(_asan_enabled_)
+    Y_UNUSED(size);
+    return ptr;
+#endif
+}
+
+} // namespace NKikimr
diff --git a/yql/essentials/minikql/comp_nodes/mkql_grace_join.cpp b/yql/essentials/minikql/comp_nodes/mkql_grace_join.cpp
index bb319984526..1ed5ff6bb9d 100644
--- a/yql/essentials/minikql/comp_nodes/mkql_grace_join.cpp
+++ b/yql/essentials/minikql/comp_nodes/mkql_grace_join.cpp
@@ -202,6 +202,8 @@ void TGraceJoinPacker::Pack()  {
 
     TuplesPacked++;
     std::fill(TupleIntVals.begin(), TupleIntVals.end(), 0);
+    std::fill(TupleStrings.begin(), TupleStrings.end(), nullptr);
+    std::fill(TupleStrSizes.begin(), TupleStrSizes.end(), 0);
 
     for (ui64 i = 0; i < ColumnsPackInfo.size(); i++) {
 
diff --git a/yql/essentials/minikql/comp_nodes/ut/mkql_wide_combine_ut.cpp b/yql/essentials/minikql/comp_nodes/ut/mkql_wide_combine_ut.cpp
index 49e292d8eb9..3affc77478a 100644
--- a/yql/essentials/minikql/comp_nodes/ut/mkql_wide_combine_ut.cpp
+++ b/yql/essentials/minikql/comp_nodes/ut/mkql_wide_combine_ut.cpp
@@ -496,6 +496,8 @@ Y_UNIT_TEST_SUITE(TMiniKQLWideCombinerTest) {
         UNIT_ASSERT_EQUAL(streamVal.Fetch(result), NUdf::EFetchStatus::Finish);
     }
 
+// Do not run under ASAN since memory limits is hard to track. Every allocation produce and tracks more memory than requested.
+#if !defined(_asan_enabled_)
     Y_UNIT_TEST_LLVM(TestSkipYieldRespectsMemLimit) {
         TTestStreamParams params;
         params.StringSize = 50000;
@@ -528,6 +530,7 @@ Y_UNIT_TEST_SUITE(TMiniKQLWideCombinerTest) {
         UNIT_ASSERT_EQUAL(streamVal.Fetch(result), NUdf::EFetchStatus::Finish);
         UNIT_ASSERT_EQUAL(streamVal.Fetch(result), NUdf::EFetchStatus::Finish);
     }
+#endif // defined(_asan_enabled_)
 }
 
 Y_UNIT_TEST_SUITE(TMiniKQLWideCombinerPerfTest) {
diff --git a/yql/essentials/minikql/compact_hash.h b/yql/essentials/minikql/compact_hash.h
index ec810efe5bc..3704907d9c3 100644
--- a/yql/essentials/minikql/compact_hash.h
+++ b/yql/essentials/minikql/compact_hash.h
@@ -3,6 +3,7 @@
 #include <yql/essentials/utils/hash.h>
 
 #include "aligned_page_pool.h"
+#include "asan_utils.h"
 #include "primes.h"
 
 #include <util/generic/vector.h>
@@ -563,7 +564,7 @@ private:
         }
         ui16 listCount = GetSmallPageCapacity<T>(size);
         Y_ASSERT(listCount >= 2);
-        TListHeader* header = new (GetPagePool().GetPage()) TListHeader(SMALL_MARK, size, listCount);
+        TListHeader* header = new (SanitizerMarkValid(GetPagePool().GetPage(), TAlignedPagePool::POOL_PAGE_SIZE)) TListHeader(SMALL_MARK, size, listCount);
         pages.PushFront(&header->ListItem);
         return header;
     }
@@ -580,14 +581,14 @@ private:
         ui16 listCapacity = FastClp2(size);
         ui16 listCount = GetMediumPageCapacity<T>(listCapacity);
         Y_ASSERT(listCount >= 2);
-        TListHeader* header = new (GetPagePool().GetPage()) TListHeader(MEDIUM_MARK, listCapacity, listCount);
+        TListHeader* header = new (SanitizerMarkValid(GetPagePool().GetPage(), TAlignedPagePool::POOL_PAGE_SIZE)) TListHeader(MEDIUM_MARK, listCapacity, listCount);
         pages.PushFront(&header->ListItem);
         return header;
     }
 
     template <typename T>
     TLargeListHeader* GetLargeListPage() {
-        TLargeListHeader* const header = new (GetPagePool().GetPage()) TLargeListHeader(GetLargePageCapacity<T>());
+        TLargeListHeader* const header = new (SanitizerMarkValid(GetPagePool().GetPage(), TAlignedPagePool::POOL_PAGE_SIZE)) TLargeListHeader(GetLargePageCapacity<T>());
         return header;
     }
 
@@ -1315,7 +1316,7 @@ protected:
 
     void AllocateBuckets(size_t count) {
         auto bucketsMemory = Max(sizeof(TItemNode) * count, (size_t)TAlignedPagePool::POOL_PAGE_SIZE);
-        Buckets_ = (TItemNode*)GetPagePool().GetBlock(bucketsMemory);
+        Buckets_ = (TItemNode*)SanitizerMarkValid(GetPagePool().GetBlock(bucketsMemory), bucketsMemory);
         BucketsCount_ = count;
         BucketsMemory_ = bucketsMemory;
         for (size_t i = 0; i < count; ++i) {
diff --git a/yql/essentials/minikql/gtest_ut/allocator_ut.cpp b/yql/essentials/minikql/gtest_ut/allocator_ut.cpp
new file mode 100644
index 00000000000..f4ae5a02926
--- /dev/null
+++ b/yql/essentials/minikql/gtest_ut/allocator_ut.cpp
@@ -0,0 +1,168 @@
+#include <yql/essentials/minikql/mkql_alloc.cpp>
+
+#include <library/cpp/testing/gtest/gtest.h>
+
+#include <cstdlib>
+
+namespace NKikimr::NMiniKQL {
+
+enum class EAllocatorType {
+    DefaultAllocator,
+    ArrowAllocator,
+    HugeAllocator,
+};
+
+class MemoryTest: public ::testing::TestWithParam<std::tuple<int, EAllocatorType>> {
+protected:
+    MemoryTest()
+        : ScopedAlloc_(__LOCATION__) {
+    }
+
+    size_t AllocSize() const {
+        return static_cast<size_t>(std::get<0>(GetParam()));
+    }
+
+    EAllocatorType GetAllocatorType() const {
+        return std::get<1>(GetParam());
+    }
+
+    void* AllocateMemory(size_t size) const {
+        EAllocatorType allocatorType = GetAllocatorType();
+        switch (allocatorType) {
+            case EAllocatorType::DefaultAllocator:
+                return TWithDefaultMiniKQLAlloc::AllocWithSize(size);
+            case EAllocatorType::ArrowAllocator:
+                return MKQLArrowAllocate(size);
+            case EAllocatorType::HugeAllocator:
+                return TMKQLHugeAllocator<char>::allocate(size);
+            default:
+                return nullptr; // Should never reach here
+        }
+    }
+
+    void Free(const void* mem, size_t size) const {
+        EAllocatorType allocatorType = GetAllocatorType();
+        switch (allocatorType) {
+            case EAllocatorType::DefaultAllocator:
+                TWithDefaultMiniKQLAlloc::FreeWithSize(mem, size);
+                break;
+            case EAllocatorType::ArrowAllocator:
+                MKQLArrowFree(mem, size);
+                break;
+            case EAllocatorType::HugeAllocator:
+                TMKQLHugeAllocator<char>::deallocate(const_cast<char*>(static_cast<const char*>(mem)), size);
+                break;
+            default:
+                break; // Should never reach here
+        }
+    }
+
+    void AccessMemory(volatile void* memory, ssize_t offset) const {
+        volatile char* ptr = static_cast<volatile char*>(memory) + offset;
+        *ptr = 'A'; // Perform a basic write operation
+    }
+
+private:
+    TScopedAlloc ScopedAlloc_;
+};
+
+// Test naming function
+std::string TestNameGenerator(const ::testing::TestParamInfo<MemoryTest::ParamType>& info) {
+    int sizeNumber = std::get<0>(info.param);
+    EAllocatorType allocatorType = std::get<1>(info.param);
+
+
+    std::string allocatorName = [&] () {
+        switch (allocatorType) {
+            case EAllocatorType::DefaultAllocator:
+                return "DefaultAllocator";
+            case EAllocatorType::ArrowAllocator:
+                return "ArrowAllocator";
+            case EAllocatorType::HugeAllocator:
+                return "HugeAllocator";
+        }
+    }();
+
+    return "Size" + std::to_string(sizeNumber) + "With" + allocatorName + "Allocator";
+}
+
+// Out of bounds access + use after free can be tested only with
+// --sanitize=address.
+#if defined(_asan_enabled_)
+TEST_P(MemoryTest, AccessOutOfBounds) {
+    size_t allocationSize = AllocSize();
+
+    void* memory = AllocateMemory(allocationSize);
+    ASSERT_NE(memory, nullptr) << "Memory allocation failed.";
+    // Accessing valid memory.
+    ASSERT_NO_THROW({
+        AccessMemory(memory, 0);
+        AccessMemory(memory, allocationSize - 1);
+    });
+
+    // Accessing invalid left memory.
+    EXPECT_DEATH({ AccessMemory(memory, -1); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, -8); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, -16); }, "");
+
+    // Accessing invalid right memory.
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize + 6); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize + 12); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize + 15); }, "");
+
+    Free(memory, allocationSize);
+}
+
+TEST_P(MemoryTest, AccessAfterFree) {
+    size_t allocationSize = AllocSize();
+    void* memory = AllocateMemory(allocationSize);
+    void* memory2 = AllocateMemory(allocationSize);
+    ASSERT_NE(memory, nullptr) << "Memory allocation failed.";
+    Free(memory, allocationSize);
+
+    // Access after free — should crash
+    EXPECT_DEATH({ AccessMemory(memory, 0); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize / 2); }, "");
+    EXPECT_DEATH({ AccessMemory(memory, allocationSize - 1); }, "");
+
+    Free(memory2, allocationSize);
+    // Access after free — should crash
+    EXPECT_DEATH({ AccessMemory(memory2, 0); }, "");
+    EXPECT_DEATH({ AccessMemory(memory2, allocationSize / 2); }, "");
+    EXPECT_DEATH({ AccessMemory(memory2, allocationSize - 1); }, "");
+}
+
+#endif // defined(_asan_enabled_)
+
+// Double free tracked only in DEBUG mode.
+#ifndef NDEBUG
+TEST_P(MemoryTest, DoubleFree) {
+    if (GetAllocatorType() == EAllocatorType::ArrowAllocator || GetAllocatorType() == EAllocatorType::HugeAllocator) {
+        GTEST_SKIP() << "Arrow and Huge allocators arae not instrumented yet to track double free.";
+    }
+    size_t allocationSize = AllocSize();
+
+    void* memory = AllocateMemory(allocationSize);
+    ASSERT_NE(memory, nullptr) << "Memory allocation failed.";
+
+    Free(memory, allocationSize);
+
+    // Attempting double free — should crash
+    EXPECT_DEATH({ Free(memory, allocationSize); }, "");
+}
+#endif // NDEBUG
+
+// Allow empty tests for MSAN and other sanitizers.
+GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(MemoryTest);
+
+INSTANTIATE_TEST_SUITE_P(MemoryTests, MemoryTest,
+                         ::testing::Combine(
+                             ::testing::Values(8, 64, 32 * 1024, 64 * 1024, 128 * 1024, 64 * 1024 * 1024),
+                             ::testing::Values(
+                                 EAllocatorType::DefaultAllocator,
+                                 EAllocatorType::ArrowAllocator,
+                                 EAllocatorType::HugeAllocator)),
+                         TestNameGenerator);
+
+} // namespace NKikimr::NMiniKQL
diff --git a/yql/essentials/minikql/gtest_ut/ya.make b/yql/essentials/minikql/gtest_ut/ya.make
new file mode 100644
index 00000000000..ed80e31b208
--- /dev/null
+++ b/yql/essentials/minikql/gtest_ut/ya.make
@@ -0,0 +1,17 @@
+GTEST()
+
+PEERDIR(
+    yql/essentials/minikql
+    yql/essentials/minikql/invoke_builtins/llvm16
+    yql/essentials/public/udf/service/exception_policy
+    contrib/libs/apache/arrow
+    yql/essentials/sql/pg_dummy
+)
+
+SRC(
+    allocator_ut.cpp
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()
diff --git a/yql/essentials/minikql/mkql_alloc.cpp b/yql/essentials/minikql/mkql_alloc.cpp
index 27a963380e9..a8933de2149 100644
--- a/yql/essentials/minikql/mkql_alloc.cpp
+++ b/yql/essentials/minikql/mkql_alloc.cpp
@@ -1,6 +1,10 @@
 #include "mkql_alloc.h"
-#include <util/system/align.h>
+
 #include <yql/essentials/public/udf/udf_value.h>
+
+#include <util/system/align.h>
+#include <util/generic/scope.h>
+
 #include <tuple>
 
 namespace NKikimr {
@@ -204,6 +208,7 @@ void* MKQLAllocSlow(size_t sz, TAllocState* state, const EMemorySubPool mPool) {
     auto roundedSize = AlignUp(sz + sizeof(TAllocPageHeader), MKQL_ALIGNMENT);
     auto capacity = Max(ui64(TAlignedPagePool::POOL_PAGE_SIZE), roundedSize);
     auto currPage = (TAllocPageHeader*)state->GetBlock(capacity);
+    SanitizerMarkValid(currPage, sizeof(TAllocPageHeader));
     currPage->Deallocated = 0;
     currPage->Capacity = capacity;
     currPage->Offset = roundedSize;
@@ -239,6 +244,7 @@ void* TPagedArena::AllocSlow(const size_t sz, const EMemorySubPool mPool) {
     auto roundedSize = AlignUp(sz + sizeof(TAllocPageHeader), MKQL_ALIGNMENT);
     auto capacity = Max(ui64(TAlignedPagePool::POOL_PAGE_SIZE), roundedSize);
     currentPage = (TAllocPageHeader*)PagePool_->GetBlock(capacity);
+    SanitizerMarkValid(currentPage, sizeof(TAllocPageHeader));
     currentPage->Capacity = capacity;
     void* ret = (char*)currentPage + sizeof(TAllocPageHeader);
     currentPage->Offset = roundedSize;
@@ -267,7 +273,6 @@ void* MKQLArrowAllocateOnArena(ui64 size) {
 
     auto alignedSize = AlignUp(size, ArrowAlignment);
     auto& page = state->CurrentArrowPages;
-
     if (Y_UNLIKELY(!page || page->Offset + alignedSize > page->Size)) {
         const auto pageSize = TAllocState::POOL_PAGE_SIZE;
 
@@ -280,6 +285,7 @@ void* MKQLArrowAllocateOnArena(ui64 size) {
         }
 
         page = (TMkqlArrowHeader*)GetAlignedPage();
+        SanitizerMarkValid(page, sizeof(TMkqlArrowHeader));
         page->Offset = sizeof(TMkqlArrowHeader);
         page->Size = pageSize;
         page->UseCount = 1;
@@ -295,11 +301,11 @@ void* MKQLArrowAllocateOnArena(ui64 size) {
     void* ptr = (ui8*)page + page->Offset;
     page->Offset += alignedSize;
     ++page->UseCount;
-
     return ptr;
 }
 
-void* MKQLArrowAllocate(ui64 size) {
+namespace {
+void* MKQLArrowAllocateImpl(ui64 size) {
     if (Y_LIKELY(!TAllocState::IsDefaultAllocatorUsed())) {
         if (size <= ArrowSizeForArena) {
             return MKQLArrowAllocateOnArena(size);
@@ -324,6 +330,7 @@ void* MKQLArrowAllocate(ui64 size) {
     }
 
     auto* header = (TMkqlArrowHeader*)ptr;
+    SanitizerMarkValid(header, sizeof(TMkqlArrowHeader));
     header->Offset = 0;
     header->UseCount = 0;
 
@@ -337,6 +344,13 @@ void* MKQLArrowAllocate(ui64 size) {
     header->Size = size;
     return header + 1;
 }
+} // namespace
+
+void* MKQLArrowAllocate(ui64 size) {
+    auto sizeWithRedzones = GetSizeToAlloc(size);
+    void* mem = MKQLArrowAllocateImpl(sizeWithRedzones);
+    return WrapPointerWithRedZones(mem, sizeWithRedzones);
+}
 
 void* MKQLArrowReallocate(const void* mem, ui64 prevSize, ui64 size) {
     auto res = MKQLArrowAllocate(size);
@@ -358,14 +372,15 @@ void MKQLArrowFreeOnArena(const void* ptr) {
             Y_ENSURE(it != state->ArrowBuffers.end());
             state->ArrowBuffers.erase(it);
         }
-
+        SanitizerMarkInvalid(page, sizeof(TMkqlArrowHeader));
         ReleaseAlignedPage(page);
     }
 
     return;
 }
 
-void MKQLArrowFree(const void* mem, ui64 size) {
+namespace {
+void MKQLArrowFreeImpl(const void* mem, ui64 size) {
     if (Y_LIKELY(!TAllocState::IsDefaultAllocatorUsed())) {
         if (size <= ArrowSizeForArena) {
             return MKQLArrowFreeOnArena(mem);
@@ -393,8 +408,16 @@ void MKQLArrowFree(const void* mem, ui64 size) {
 
     ReleaseAlignedPage(header, fullSize);
 }
+} // namespace
+
+void MKQLArrowFree(const void* mem, ui64 size) {
+    mem = UnwrapPointerWithRedZones(mem, size);
+    auto sizeWithRedzones = GetSizeToAlloc(size);
+    return MKQLArrowFreeImpl(mem, sizeWithRedzones);
+}
 
 void MKQLArrowUntrack(const void* mem, ui64 size) {
+    mem = GetOriginalAllocatedObject(mem, size);
     TAllocState* state = TlsAllocState;
     Y_ENSURE(state);
     if (!state->EnableArrowTracking) {
diff --git a/yql/essentials/minikql/mkql_alloc.h b/yql/essentials/minikql/mkql_alloc.h
index 94a77471065..7a7716dd659 100644
--- a/yql/essentials/minikql/mkql_alloc.h
+++ b/yql/essentials/minikql/mkql_alloc.h
@@ -1,23 +1,26 @@
 #pragma once
+
 #include "aligned_page_pool.h"
 #include "mkql_mem_info.h"
+
 #include <yql/essentials/core/pg_settings/guc_settings.h>
+#include <yql/essentials/minikql/asan_utils.h>
 #include <yql/essentials/parser/pg_wrapper/interface/context.h>
 #include <yql/essentials/public/udf/udf_allocator.h>
 #include <yql/essentials/public/udf/udf_value.h>
+
 #include <util/string/builder.h>
 #include <util/system/align.h>
 #include <util/system/defaults.h>
 #include <util/system/tls.h>
-#include <new>
+#include <util/generic/scope.h>
+
 #include <unordered_map>
 #include <atomic>
 #include <memory>
 #include <source_location>
 
-namespace NKikimr {
-
-namespace NMiniKQL {
+namespace NKikimr::NMiniKQL {
 
 const ui64 MKQL_ALIGNMENT = 16;
 
@@ -116,7 +119,7 @@ struct TAllocState : public TAlignedPagePool
     explicit TAllocState(const TSourceLocation& location, const TAlignedPagePoolCounters& counters, bool supportsSizedAllocators);
     void KillAllBoxed();
     void InvalidateMemInfo();
-    size_t GetDeallocatedInPages() const;
+    Y_NO_SANITIZE("address") size_t GetDeallocatedInPages() const;
     static void CleanupPAllocList(TListEntry* root);
     static void CleanupArrowList(TListEntry* root);
 
@@ -284,17 +287,22 @@ public:
         Clear();
     }
 
-    void* Alloc(size_t sz, const EMemorySubPool pagePool = EMemorySubPool::Default) {
+    void* AllocImpl(size_t sz, const EMemorySubPool pagePool) {
         auto& currentPage = CurrentPages_[(TMemorySubPoolIdx)pagePool];
         if (Y_LIKELY(currentPage->Offset + sz <= currentPage->Capacity)) {
             void* ret = (char*)currentPage + currentPage->Offset;
             currentPage->Offset = AlignUp(currentPage->Offset + sz, MKQL_ALIGNMENT);
             return ret;
         }
-
         return AllocSlow(sz, pagePool);
     }
 
+    void* Alloc(size_t sz, const EMemorySubPool pagePool = EMemorySubPool::Default) {
+        sz = GetSizeToAlloc(sz);
+        void* mem = AllocImpl(sz, pagePool);
+        return WrapPointerWithRedZones(mem, sz);
+    }
+
     void Clear() noexcept;
 
 private:
@@ -344,7 +352,7 @@ inline void* MKQLAllocFastDeprecated(size_t sz, TAllocState* state, const EMemor
     return ret;
 }
 
-inline void* MKQLAllocFastWithSize(size_t sz, TAllocState* state, const EMemorySubPool mPool, const TAllocLocation& location = TAllocLocation::current()) {
+inline void* MKQLAllocFastWithSizeImpl(size_t sz, TAllocState* state, const EMemorySubPool mPool, const TAllocLocation& location) {
 #ifdef NDEBUG
     Y_UNUSED(location);
 #endif
@@ -384,6 +392,12 @@ inline void* MKQLAllocFastWithSize(size_t sz, TAllocState* state, const EMemoryS
     return ret;
 }
 
+inline void* MKQLAllocFastWithSize(size_t sz, TAllocState* state, const EMemorySubPool mPool, const TAllocLocation& location = TAllocLocation::current()) {
+    sz = GetSizeToAlloc(sz);
+    void* mem = MKQLAllocFastWithSizeImpl(sz, state, mPool, location);
+    return WrapPointerWithRedZones(mem, sz);
+}
+
 void MKQLFreeSlow(TAllocPageHeader* header, TAllocState *state, const EMemorySubPool mPool) noexcept;
 
 inline void MKQLFreeDeprecated(const void* mem, const EMemorySubPool mPool) noexcept {
@@ -415,7 +429,7 @@ inline void MKQLFreeDeprecated(const void* mem, const EMemorySubPool mPool) noex
     MKQLFreeSlow(header, TlsAllocState, mPool);
 }
 
-inline void MKQLFreeFastWithSize(const void* mem, size_t sz, TAllocState* state, const EMemorySubPool mPool) noexcept {
+inline void MKQLFreeFastWithSizeImpl(const void* mem, size_t sz, TAllocState* state, const EMemorySubPool mPool) noexcept {
     if (!mem) {
         return;
     }
@@ -436,18 +450,26 @@ inline void MKQLFreeFastWithSize(const void* mem, size_t sz, TAllocState* state,
     }
 
     TAllocPageHeader* header = (TAllocPageHeader*)TAllocState::GetPageStart(mem);
-    Y_DEBUG_ABORT_UNLESS(header->MyAlloc == state, "%s", (TStringBuilder() << "wrong allocator was used; "
-        "allocated with: " << header->MyAlloc->GetDebugInfo() << " freed with: " << TlsAllocState->GetDebugInfo()).data());
-    if (Y_LIKELY(--header->UseCount != 0)) {
-        header->Deallocated += sz;
-        return;
+    {
+        Y_DEBUG_ABORT_UNLESS(header->MyAlloc == state, "Wrong allocator was used. Allocated with: %s, freed with: %s",
+                             header->MyAlloc->GetDebugInfo().c_str(), TlsAllocState->GetDebugInfo().c_str());
+        if (Y_LIKELY(--header->UseCount != 0)) {
+            header->Deallocated += sz;
+            return;
+        }
     }
 
     MKQLFreeSlow(header, state, mPool);
 }
 
-inline void* MKQLAllocDeprecated(size_t sz, const EMemorySubPool mPool, const TAllocLocation& location = TAllocLocation::current()) {
-    return MKQLAllocFastDeprecated(sz, TlsAllocState, mPool, location);
+inline void MKQLFreeFastWithSize(const void* mem, size_t sz, TAllocState* state, const EMemorySubPool mPool) noexcept {
+    mem = UnwrapPointerWithRedZones(mem, sz);
+    sz = GetSizeToAlloc(sz);
+    return MKQLFreeFastWithSizeImpl(mem, sz, state, mPool);
+}
+
+inline void* MKQLAllocDeprecated(size_t sz, const EMemorySubPool mPool) {
+    return MKQLAllocFastDeprecated(sz, TlsAllocState, mPool);
 }
 
 inline void* MKQLAllocWithSize(size_t sz, const EMemorySubPool mPool, const TAllocLocation& location = TAllocLocation::current()) {
@@ -568,17 +590,31 @@ struct TMKQLHugeAllocator
     template<typename U> bool operator==(const TMKQLHugeAllocator<U>&) const { return true; }
     template<typename U> bool operator!=(const TMKQLHugeAllocator<U>&) const { return false; }
 
-    static pointer allocate(size_type n, const void* = nullptr)
+    static pointer allocateImpl(size_type n, const void* = nullptr)
     {
         size_t size = Max(n * sizeof(value_type), TAllocState::POOL_PAGE_SIZE);
         return static_cast<pointer>(TlsAllocState->GetBlock(size));
     }
 
-    static void deallocate(const_pointer p, size_type n) noexcept
+    static pointer allocate(size_type n, const void* = nullptr)
+    {
+        n = GetSizeToAlloc(n);
+        void* mem = allocateImpl(n);
+        return static_cast<pointer>(WrapPointerWithRedZones(mem, n));
+    }
+
+    static void deallocateImpl(const_pointer p, size_type n) noexcept
     {
         size_t size = Max(n * sizeof(value_type), TAllocState::POOL_PAGE_SIZE);
         TlsAllocState->ReturnBlock(const_cast<pointer>(p), size);
     }
+
+    static void deallocate(const_pointer p, size_type n) noexcept
+    {
+        p = static_cast<const_pointer>(UnwrapPointerWithRedZones(p, n));
+        n = GetSizeToAlloc(n);
+        return deallocateImpl(p, n);
+    }
 };
 
 template <typename T>
@@ -611,7 +647,7 @@ public:
             return;
         }
 
-        auto ptr = Pool.GetPage();
+        auto ptr = SanitizerMarkValid(Pool.GetPage(), TAlignedPagePool::POOL_PAGE_SIZE);
         IndexInLastPage = 1;
         Pages.push_back(ptr);
         new(ptr) T(std::move(value));
@@ -808,6 +844,4 @@ inline void TBoxedValueWithFree::operator delete(void *mem) noexcept {
     MKQLFreeWithSize(mem, size, EMemorySubPool::Default);
 }
 
-} // NMiniKQL
-
-} // NKikimr
+} // namespace NKikimr::NMiniKQL
diff --git a/yql/essentials/minikql/mkql_alloc_ut.cpp b/yql/essentials/minikql/mkql_alloc_ut.cpp
index 9dfd726e7ad..c1636375c35 100644
--- a/yql/essentials/minikql/mkql_alloc_ut.cpp
+++ b/yql/essentials/minikql/mkql_alloc_ut.cpp
@@ -31,6 +31,12 @@ Y_UNIT_TEST_SUITE(TMiniKQLAllocTest) {
 
     Y_UNIT_TEST(TestDeallocated) {
         TScopedAlloc alloc(__LOCATION__);
+#if defined(_asan_enabled_)
+        constexpr size_t EXTRA_ALLOCATION_SPACE = ASAN_EXTRA_ALLOCATION_SPACE;
+#else  // defined(_asan_enabled_)
+        constexpr size_t EXTRA_ALLOCATION_SPACE = 0;
+#endif // defined(_asan_enabled_)
+
         void* p1 = TWithDefaultMiniKQLAlloc::AllocWithSize(10);
         void* p2 = TWithDefaultMiniKQLAlloc::AllocWithSize(20);
         UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetUsed(), TAlignedPagePool::POOL_PAGE_SIZE);
@@ -38,7 +44,7 @@ Y_UNIT_TEST_SUITE(TMiniKQLAllocTest) {
         UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetFreePageCount(), 0);
         TWithDefaultMiniKQLAlloc::FreeWithSize(p1, 10);
         UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetUsed(), TAlignedPagePool::POOL_PAGE_SIZE);
-        UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetDeallocatedInPages(), 10);
+        UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetDeallocatedInPages(), 10 + EXTRA_ALLOCATION_SPACE);
         UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetFreePageCount(), 0);
         TWithDefaultMiniKQLAlloc::FreeWithSize(p2, 20);
         UNIT_ASSERT_VALUES_EQUAL(alloc.Ref().GetUsed(), 0);
diff --git a/yql/essentials/minikql/mkql_mem_info.cpp b/yql/essentials/minikql/mkql_mem_info.cpp
index a8441d26873..5628531aaa9 100644
--- a/yql/essentials/minikql/mkql_mem_info.cpp
+++ b/yql/essentials/minikql/mkql_mem_info.cpp
@@ -67,9 +67,8 @@ void TMemoryUsageInfo::Return(const void* mem, ui64 size) {
         Y_DEBUG_ABORT_UNLESS(it != AllocationsMap_.end(), "Double free at: %p", mem);
     }
 
-    Y_DEBUG_ABORT_UNLESS(size == it->second.Size,
-                "Deallocating wrong size at: %p, "
-                "allocated at: %s", mem, (TStringBuilder() << it->second.Location).c_str());
+    Y_DEBUG_ABORT_UNLESS(size == it->second.Size, "Deallocating wrong size at: %p, allocated at: %s. Actual size: %zu, but expected size is: %zu",
+                         mem, (TStringBuilder() << it->second.Location).c_str(), size, it->second.Size);
     if (AllowMissing_) {
         it->second.IsDeleted = true;
     } else {
@@ -141,4 +140,4 @@ void TMemoryUsageInfo::VerifyDebug() const {
 }
 
 }
-}
\ No newline at end of file
+}
diff --git a/yql/essentials/minikql/mkql_string_util.cpp b/yql/essentials/minikql/mkql_string_util.cpp
index 78adacc231d..e9604c06924 100644
--- a/yql/essentials/minikql/mkql_string_util.cpp
+++ b/yql/essentials/minikql/mkql_string_util.cpp
@@ -1,5 +1,7 @@
 #include "mkql_string_util.h"
 
+#include <util/generic/scope.h>
+
 namespace NKikimr {
 namespace NMiniKQL {
 
@@ -31,19 +33,26 @@ NUdf::TUnboxedValuePod AppendString(const NUdf::TUnboxedValuePod value, const NU
         return result;
     } else {
         if (value.IsString()) {
-            auto str = value.AsStringValue();
-            const ui32 offset = ref.Data() - str.Data();
-            if (str.Size() == valueRef.Size() + offset) {
-                if (str.TryExpandOn(ref.Size())) {
-                    std::memcpy(str.Data() + offset + valueRef.Size(), ref.Data(), ref.Size());
-                    return NUdf::TUnboxedValuePod(std::move(str), newSize, offset);
+            auto str = value.AsRawStringValue();
+            const char* strData = str->Data();
+            const char* refData = ref.Data();
+            // Check if ref.Data() is within the memory range of str
+            if (refData >= strData && refData < strData + str->Size()) {
+                const ui32 offset = refData - strData;
+                if (str->Size() == valueRef.Size() + offset) {
+                    if (str->TryExpandOn(ref.Size())) {
+                        std::memcpy(str->Data() + offset + valueRef.Size(), ref.Data(), ref.Size());
+                        return NUdf::TUnboxedValuePod(NYql::NUdf::TStringValue(std::move(str)), newSize, offset);
+                    }
                 }
             }
         }
-
         auto data = NUdf::TStringValue::AllocateData(newSize, newSize + newSize / 2);
         NUdf::TStringValue str(data);
-        data->UnRef();
+        Y_DEFER {
+            data->ReleaseRef();
+            value.DeleteUnreferenced();
+        };
         std::memcpy(str.Data(), valueRef.Data(), valueRef.Size());
         std::memcpy(str.Data() + valueRef.Size(), ref.Data(), ref.Size());
         return NUdf::TUnboxedValuePod(std::move(str));
@@ -69,10 +78,12 @@ NUdf::TUnboxedValuePod PrependString(const NUdf::TStringRef ref, const NUdf::TUn
     } else {
         auto data = NUdf::TStringValue::AllocateData(newSize, newSize + newSize / 2);
         NUdf::TStringValue str(data);
-        data->UnRef();
+        Y_DEFER {
+            data->ReleaseRef();
+            value.DeleteUnreferenced();
+        };
         std::memcpy(str.Data(), ref.Data(), ref.Size());
         std::memcpy(str.Data() + ref.Size(), valueRef.Data(), valueRef.Size());
-        value.DeleteUnreferenced();
         return NUdf::TUnboxedValuePod(std::move(str));
     }
 }
@@ -96,23 +107,31 @@ NUdf::TUnboxedValuePod ConcatStrings(const NUdf::TUnboxedValuePod first, const N
         return result;
     } else {
         if (first.IsString()) {
-            auto str = first.AsStringValue();
-            const ui32 offset = leftRef.Data() - str.Data();
-            if (str.Size() == leftRef.Size() + offset) {
-                if (str.TryExpandOn(rightRef.Size())) {
-                    std::memcpy(str.Data() + offset + leftRef.Size(), rightRef.Data(), rightRef.Size());
-                    second.DeleteUnreferenced();
-                    return NUdf::TUnboxedValuePod(std::move(str), newSize, offset);
+            auto str = first.AsRawStringValue();
+            const char* strData = str->Data();
+            const char* leftRefData = leftRef.Data();
+            // Check if leftRef.Data() is within the memory range of str
+            if (leftRefData >= strData && leftRefData < strData + str->Size()) {
+                const ui32 offset = leftRefData - strData;
+                if (str->Size() == leftRef.Size() + offset) {
+                    if (str->TryExpandOn(rightRef.Size())) {
+                        std::memcpy(str->Data() + offset + leftRef.Size(), rightRef.Data(), rightRef.Size());
+                        second.DeleteUnreferenced();
+                        return NUdf::TUnboxedValuePod(NUdf::TStringValue(str), newSize, offset);
+                    }
                 }
             }
         }
 
         auto data = NUdf::TStringValue::AllocateData(newSize, newSize + newSize / 2);
         NUdf::TStringValue str(data);
-        data->UnRef();
+        Y_DEFER {
+            data->ReleaseRef();
+            second.DeleteUnreferenced();
+            first.DeleteUnreferenced();
+        };
         std::memcpy(str.Data(), leftRef.Data(), leftRef.Size());
         std::memcpy(str.Data() + leftRef.Size(), rightRef.Data(), rightRef.Size());
-        second.DeleteUnreferenced();
         return NUdf::TUnboxedValuePod(std::move(str));
     }
 }
@@ -134,13 +153,22 @@ NUdf::TUnboxedValuePod SubString(const NUdf::TUnboxedValuePod value, ui32 offset
         value.DeleteUnreferenced();
         return result;
     } else {
-        auto old = value.AsStringValue();
-        if (const auto newOffset = ui32(ref.Data() - old.Data()) + offset; NUdf::TUnboxedValuePod::OffsetLimit > newOffset)
-            return NUdf::TUnboxedValuePod(std::move(old), newSize, newOffset);
+        auto old = value.AsRawStringValue();
+        const char* oldData = old->Data();
+        const char* refData = ref.Data();
+        // Check if ref.Data() is within the memory range of old
+        if (refData >= oldData && refData < oldData + old->Size()) {
+            if (const auto newOffset = ui32(refData - oldData) + offset; NUdf::TUnboxedValuePod::OffsetLimit > newOffset) {
+                return NUdf::TUnboxedValuePod(NUdf::TStringValue(old), newSize, newOffset);
+            }
+        }
 
         auto data = NUdf::TStringValue::AllocateData(newSize, newSize + (newSize >> 1U));
         NUdf::TStringValue str(data);
-        data->UnRef();
+        Y_DEFER {
+            data->ReleaseRef();
+            value.DeleteUnreferenced();
+        };
         std::memcpy(str.Data(), ref.Data() + offset, newSize);
         return NUdf::TUnboxedValuePod(std::move(str));
     }
diff --git a/yql/essentials/minikql/ya.make b/yql/essentials/minikql/ya.make
index 4c09b76c66f..fe830919d02 100644
--- a/yql/essentials/minikql/ya.make
+++ b/yql/essentials/minikql/ya.make
@@ -108,4 +108,5 @@ RECURSE(
 RECURSE_FOR_TESTS(
     benchmark
     ut
+    gtest_ut
 )
diff --git a/yql/essentials/parser/pg_wrapper/comp_factory.cpp b/yql/essentials/parser/pg_wrapper/comp_factory.cpp
index 8c833132baf..c992daf8475 100644
--- a/yql/essentials/parser/pg_wrapper/comp_factory.cpp
+++ b/yql/essentials/parser/pg_wrapper/comp_factory.cpp
@@ -107,10 +107,8 @@ extern void MkqlDelete(MemoryContext context);
 extern MemoryContext MkqlGetChunkContext(void *pointer);
 extern Size MkqlGetChunkSpace(void *pointer);
 extern bool MkqlIsEmpty(MemoryContext context);
-extern void MkqlStats(MemoryContext context,
-						  MemoryStatsPrintFunc printfunc, void *passthru,
-						      MemoryContextCounters *totals,
-						  bool print_to_stderr);
+extern void MkqlStats(MemoryContext context, MemoryStatsPrintFunc printfunc, void* passthru, MemoryContextCounters* totals,
+                      bool print_to_stderr);
 #ifdef MEMORY_CONTEXT_CHECKING
 extern void MkqlCheck(MemoryContext context);
 #endif
@@ -1337,6 +1335,11 @@ public:
         NUdf::TUnboxedValuePod res;
         if constexpr (!UseContext) {
             TPAllocScope call;
+            Y_DEFER {
+                // This ensures that there is no dangling pointers references to freed
+                // |TPAllocScope| pages that can be allocated and stored inside |callInfo.flinfo->fn_extra|.
+                callInfo.flinfo->fn_extra = nullptr;
+            };
             res = this->DoCall(callInfo);
         } else {
             res = this->DoCall(callInfo);
diff --git a/yql/essentials/public/udf/udf_string.h b/yql/essentials/public/udf/udf_string.h
index 72b2c347622..96e97a71b51 100644
--- a/yql/essentials/public/udf/udf_string.h
+++ b/yql/essentials/public/udf/udf_string.h
@@ -94,6 +94,20 @@ class TStringValue
             Refs_ = prev;
         }
 
+        bool TryExpandOn(ui32 len) {
+            if (RefCount() < 0) {
+                return false;
+            }
+
+            const auto size = Size_ + len;
+            if (Capacity_ < size) {
+                return false;
+            }
+
+            Size_ = size;
+            return true;
+        }
+
     private:
         ui32 Size_;
         i32  Refs_;
@@ -169,12 +183,7 @@ public:
         if (RefCount() < 0)
            return false;
 
-        const auto size = Data_->Size_ + len;
-        if (Data_->Capacity_ < size)
-            return false;
-
-        Data_->Size_ = size;
-        return true;
+        return Data_->TryExpandOn(len);
     }
 
     inline i32 LockRef() noexcept {
diff --git a/yql/essentials/udfs/common/datetime2/datetime_udf.cpp b/yql/essentials/udfs/common/datetime2/datetime_udf.cpp
index c159f63d5e5..35f0af4150f 100644
--- a/yql/essentials/udfs/common/datetime2/datetime_udf.cpp
+++ b/yql/essentials/udfs/common/datetime2/datetime_udf.cpp
@@ -3183,14 +3183,37 @@ private:
                             }
                             SetYear<TMResourceName>(result, year);
                         } else {
-                            static constexpr size_t size = 6;
                             i64 year = 0LL;
                             i64 negative = 1LL;
+                            if (limit == 0) {
+                                // Year must take at least 1 byte.
+                                return false;
+                            }
                             if (*it == '-') {
                                 negative = -1LL;
                                 it++;
+                                limit--;
                             }
-                            if (!ParseNDigits<size, true>::Do(it, year) || !ValidateYear<TM64ResourceName>(negative * year)) {
+                            auto parseDigits = [&]() {
+                                switch (limit) {
+                                    case 0:
+                                        // Year number must take at least 1 byte.
+                                        return false;
+                                    case 1:
+                                        return ParseNDigits<1, true>::Do(it, year);
+                                    case 2:
+                                        return ParseNDigits<2, true>::Do(it, year);
+                                    case 3:
+                                        return ParseNDigits<3, true>::Do(it, year);
+                                    case 4:
+                                        return ParseNDigits<4, true>::Do(it, year);
+                                    case 5:
+                                        return ParseNDigits<5, true>::Do(it, year);
+                                    default:
+                                        return ParseNDigits<6, true>::Do(it, year);
+                                }
+                            };
+                            if (!parseDigits() || !ValidateYear<TM64ResourceName>(negative * year)) {
                                 return false;
                             }
                             SetYear<TM64ResourceName>(result, negative * year);
diff --git a/yt/yql/providers/yt/codec/yt_codec.cpp b/yt/yql/providers/yt/codec/yt_codec.cpp
index 14e317df97f..e414aa0f6d1 100644
--- a/yt/yql/providers/yt/codec/yt_codec.cpp
+++ b/yt/yql/providers/yt/codec/yt_codec.cpp
@@ -1791,7 +1791,7 @@ public:
     void operator=(const TTempBlockWriter&) = delete;
 
     std::pair<char*, char*> NextEmptyBlock() override {
-        auto newPage = Pool_.GetPage();
+        auto newPage = SanitizerMarkValid(Pool_.GetPage(), TAlignedPagePool::POOL_PAGE_SIZE);
         auto header = (TPageHeader*)newPage;
         header->Avail_ = 0;
         header->Next_ = &Dummy_;