Intermediate changes

commit_hash:6476b453be16b7db003e75789e3b630c1168f14f

14 files changed, 544 insertions, 0 deletions

diff --git a/yql/essentials/sql/v1/complete/name/cache/byte_size.cpp b/yql/essentials/sql/v1/complete/name/cache/byte_size.cpp
new file mode 100644
index 00000000000..304a865aab1
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/byte_size.cpp
@@ -0,0 +1 @@
+#include "byte_size.h"
diff --git a/yql/essentials/sql/v1/complete/name/cache/byte_size.h b/yql/essentials/sql/v1/complete/name/cache/byte_size.h
new file mode 100644
index 00000000000..f838d364750
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/byte_size.h
@@ -0,0 +1,46 @@
+#pragma once
+
+#include <cstddef>
+#include <type_traits>
+
+#include <util/generic/vector.h>
+#include <util/generic/string.h>
+
+namespace NSQLComplete {
+
+    template <class T>
+    struct TByteSize;
+
+    template <class T>
+        requires std::is_fundamental_v<T>
+    struct TByteSize<T> {
+        size_t operator()(const T& x) const noexcept {
+            return sizeof(x);
+        }
+    };
+
+    template <class T>
+    struct TByteSize<TVector<T>> {
+        size_t operator()(const TVector<T>& x) const noexcept {
+            size_t bytes = sizeof(x);
+            bytes = Accumulate(x, bytes, [](size_t acc, const T& x) {
+                return acc + TByteSize<T>()(x);
+            });
+            bytes += x.capacity() * sizeof(T);
+            return bytes;
+        }
+    };
+
+    template <>
+    struct TByteSize<TString> {
+        size_t operator()(const TString& x) const noexcept {
+            return std::max(sizeof(x), sizeof(x) + x.capacity());
+        }
+    };
+
+    template <class T>
+    concept CByteSized = requires(const T& x) {
+        { TByteSize<T>()(x) } -> std::convertible_to<std::size_t>;
+    };
+
+} // namespace NSQLComplete
diff --git a/yql/essentials/sql/v1/complete/name/cache/cache.cpp b/yql/essentials/sql/v1/complete/name/cache/cache.cpp
new file mode 100644
index 00000000000..05b26b0cf80
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/cache.cpp
@@ -0,0 +1 @@
+#include "cache.h"
diff --git a/yql/essentials/sql/v1/complete/name/cache/cache.h b/yql/essentials/sql/v1/complete/name/cache/cache.h
new file mode 100644
index 00000000000..d65b41d5636
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/cache.h
@@ -0,0 +1,42 @@
+#pragma once
+
+#include "byte_size.h"
+
+#include <library/cpp/threading/future/future.h>
+
+#include <util/generic/ptr.h>
+#include <util/datetime/base.h>
+
+namespace NSQLComplete {
+
+    namespace NPrivate {
+
+        template <class T>
+        concept CHashable = requires(const T& x) {
+            { THash<T>()(x) } -> std::convertible_to<std::size_t>;
+        };
+
+        template <class T>
+        concept CCacheKey = std::regular<T> && CHashable<T> && CByteSized<T>;
+
+        template <class T>
+        concept CCacheValue = std::copyable<T> && CByteSized<T>;
+
+    }; // namespace NPrivate
+
+    template <NPrivate::CCacheKey TKey, NPrivate::CCacheValue TValue>
+    class ICache: public TThrRefBase {
+    public:
+        using TPtr = TIntrusivePtr<ICache>;
+
+        struct TEntry {
+            TValue Value = {};
+            bool IsExpired = true;
+        };
+
+        virtual ~ICache() = default;
+        virtual NThreading::TFuture<TEntry> Get(const TKey& key) const = 0;
+        virtual NThreading::TFuture<void> Update(const TKey& key, TValue value) const = 0;
+    };
+
+} // namespace NSQLComplete
diff --git a/yql/essentials/sql/v1/complete/name/cache/cached.cpp b/yql/essentials/sql/v1/complete/name/cache/cached.cpp
new file mode 100644
index 00000000000..7a155ac0999
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/cached.cpp
@@ -0,0 +1 @@
+#include "cached.h"
diff --git a/yql/essentials/sql/v1/complete/name/cache/cached.h b/yql/essentials/sql/v1/complete/name/cache/cached.h
new file mode 100644
index 00000000000..b265f6fde13
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/cached.h
@@ -0,0 +1,37 @@
+#pragma once
+
+#include "cache.h"
+
+namespace NSQLComplete {
+
+    template <NPrivate::CCacheKey TKey, NPrivate::CCacheValue TValue>
+    class TCachedQuery {
+    public:
+        using TFunc = std::function<NThreading::TFuture<TValue>(const TKey& key)>;
+
+        TCachedQuery(ICache<TKey, TValue>::TPtr cache, TFunc query)
+            : Cache_(std::move(cache))
+            , Query_(std::move(query))
+        {
+        }
+
+        NThreading::TFuture<TValue> operator()(TKey key) const {
+            return Cache_->Get(key).Apply([cache = Cache_,
+                                           query = Query_,
+                                           key = std::move(key)](auto f) {
+                typename ICache<TKey, TValue>::TEntry entry = f.ExtractValue();
+                if (entry.IsExpired) {
+                    query(key).Apply([cache, key = std::move(key)](auto f) {
+                        cache->Update(key, f.ExtractValue());
+                    });
+                }
+                return std::move(entry.Value);
+            });
+        }
+
+    private:
+        ICache<TKey, TValue>::TPtr Cache_;
+        TFunc Query_;
+    };
+
+} // namespace NSQLComplete
diff --git a/yql/essentials/sql/v1/complete/name/cache/cached_ut.cpp b/yql/essentials/sql/v1/complete/name/cache/cached_ut.cpp
new file mode 100644
index 00000000000..18607dea438
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/cached_ut.cpp
@@ -0,0 +1,30 @@
+#include "cached.h"
+
+#include <yql/essentials/sql/v1/complete/name/cache/local/cache.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <library/cpp/time_provider/monotonic_provider.h>
+
+using namespace NSQLComplete;
+
+Y_UNIT_TEST_SUITE(CachedQueryTests) {
+
+    Y_UNIT_TEST(OnExpired_WhenApplied_ThenDefferedUpdateAndReturnOld) {
+        size_t queried = 0;
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {.TTL = TDuration::Zero()});
+        auto cached = TCachedQuery<TString, TString>(cache, [&](const TString& key) {
+            queried += 1;
+            return NThreading::MakeFuture<TString>(key);
+        });
+        cache->Update("1", "2");
+
+        TString value = cached("1").GetValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL(value, "2");
+        UNIT_ASSERT_VALUES_EQUAL(queried, 1);
+        UNIT_ASSERT_VALUES_EQUAL(cached("1").GetValueSync(), "1");
+    }
+
+} // Y_UNIT_TEST_SUITE(CachedQueryTests)
diff --git a/yql/essentials/sql/v1/complete/name/cache/local/cache.cpp b/yql/essentials/sql/v1/complete/name/cache/local/cache.cpp
new file mode 100644
index 00000000000..05b26b0cf80
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/local/cache.cpp
@@ -0,0 +1 @@
+#include "cache.h"
diff --git a/yql/essentials/sql/v1/complete/name/cache/local/cache.h b/yql/essentials/sql/v1/complete/name/cache/local/cache.h
new file mode 100644
index 00000000000..9e4438f0022
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/local/cache.h
@@ -0,0 +1,113 @@
+#pragma once
+
+#include <yql/essentials/sql/v1/complete/name/cache/cache.h>
+
+#include <library/cpp/cache/cache.h>
+#include <library/cpp/time_provider/monotonic_provider.h>
+
+#include <util/system/mutex.h>
+
+namespace NSQLComplete {
+
+    struct TLocalCacheConfig {
+        size_t ByteCapacity = 1 * 1024 * 1024;
+        TDuration TTL = TDuration::Seconds(8);
+    };
+
+    namespace NPrivate {
+
+        template <CCacheValue TValue>
+        struct TLocalCacheCell {
+            TValue Value;
+            NMonotonic::TMonotonic Deadline;
+            size_t KeyByteSize = 0;
+            size_t CellByteSize = 0;
+        };
+
+        template <CCacheKey TKey, CCacheValue TValue>
+        class TLocalCache: public ICache<TKey, TValue> {
+        private:
+            using TEntry = ICache<TKey, TValue>::TEntry;
+            using TCell = TLocalCacheCell<TValue>;
+
+            struct TLRUSizeProvider {
+                size_t operator()(const TCell& x) noexcept {
+                    const size_t listItemContent = x.CellByteSize;
+                    const size_t listItemPtrs = sizeof(TIntrusiveListItem<void>);
+                    const size_t listItem = listItemContent + listItemPtrs;
+
+                    const size_t cacheIndexKey = x.KeyByteSize;
+                    const size_t cacheIndexListItemPtr = sizeof(void*);
+                    const size_t cacheIndexEntry = cacheIndexKey + cacheIndexListItemPtr;
+
+                    return listItem + cacheIndexEntry;
+                }
+            };
+
+            using TStorage = TLRUCache<TKey, TCell, TNoopDelete, TLRUSizeProvider>;
+
+        public:
+            TLocalCache(TIntrusivePtr<NMonotonic::IMonotonicTimeProvider> clock, TLocalCacheConfig config)
+                : Clock_(std::move(clock))
+                , Config_(std::move(config))
+                , Origin_(/* maxSize = */ Config_.ByteCapacity)
+            {
+            }
+
+            NThreading::TFuture<TEntry> Get(const TKey& key) const override {
+                TEntry entry;
+
+                with_lock (Mutex_) {
+                    if (auto it = Origin_.Find(key); it != Origin_.End()) {
+                        entry.Value = it->Value;
+                        entry.IsExpired = (it->Deadline < Clock_->Now());
+                    }
+                }
+
+                return NThreading::MakeFuture(std::move(entry));
+            }
+
+            NThreading::TFuture<void> Update(const TKey& key, TValue value) const override {
+                TCell cell = {
+                    .Value = std::move(value),
+                    .Deadline = Clock_->Now() + Config_.TTL,
+                    .KeyByteSize = TByteSize<TKey>()(key),
+                };
+
+                cell.CellByteSize =
+                    TByteSize<TValue>()(cell.Value) +
+                    sizeof(cell.Deadline) +
+                    cell.KeyByteSize;
+
+                with_lock (Mutex_) {
+                    Origin_.Update(key, std::move(cell));
+                }
+
+                return NThreading::MakeFuture();
+            }
+
+        private:
+            TIntrusivePtr<NMonotonic::IMonotonicTimeProvider> Clock_;
+            TLocalCacheConfig Config_;
+
+            TMutex Mutex_;
+            mutable TStorage Origin_;
+        };
+
+    } // namespace NPrivate
+
+    template <NPrivate::CCacheKey TKey, NPrivate::CCacheValue TValue>
+    ICache<TKey, TValue>::TPtr MakeLocalCache(
+        TIntrusivePtr<NMonotonic::IMonotonicTimeProvider> clock,
+        TLocalCacheConfig config) {
+        return new NPrivate::TLocalCache<TKey, TValue>(std::move(clock), std::move(config));
+    }
+
+    template <NPrivate::CCacheValue TValue>
+    struct TByteSize<NPrivate::TLocalCacheCell<TValue>> {
+        size_t operator()(const NPrivate::TLocalCacheCell<TValue>& x) const noexcept {
+            return x.CellByteSize;
+        }
+    };
+
+} // namespace NSQLComplete
diff --git a/yql/essentials/sql/v1/complete/name/cache/local/cache_ut.cpp b/yql/essentials/sql/v1/complete/name/cache/local/cache_ut.cpp
new file mode 100644
index 00000000000..0823991e2d4
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/local/cache_ut.cpp
@@ -0,0 +1,216 @@
+#include "cache.h"
+
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <library/cpp/time_provider/monotonic_provider.h>
+
+#include <util/random/random.h>
+#include <util/thread/pool.h>
+
+using namespace NSQLComplete;
+
+class TPausedClock: public NMonotonic::IMonotonicTimeProvider {
+public:
+    NMonotonic::TMonotonic Now() override {
+        return Now_;
+    }
+
+    void Skip(TDuration duration) {
+        Now_ += duration;
+    }
+
+private:
+    NMonotonic::TMonotonic Now_ = NMonotonic::CreateDefaultMonotonicTimeProvider()->Now();
+};
+
+struct TFat {
+    size_t Id = 0;
+
+    friend bool operator==(const TFat& lhs, const TFat& rhs) = default;
+};
+
+namespace NSQLComplete {
+
+    template <>
+    struct TByteSize<TFat> {
+        size_t operator()(const TFat&) const noexcept {
+            return 10'000;
+        }
+    };
+
+} // namespace NSQLComplete
+
+template <>
+struct THash<TFat> {
+    size_t operator()(const TFat& x) const noexcept {
+        return x.Id;
+    }
+};
+
+struct TAction {
+    bool IsGet = false;
+    TString Key = "";
+    TString Value = "";
+};
+
+TVector<TAction> GenerateRandomActions(size_t size) {
+    constexpr double GetFrequency = 0.75;
+    constexpr ui32 MaxKey = 100;
+    constexpr ui32 MinValue = 1;
+    constexpr ui32 MaxValue = 10;
+
+    TVector<TAction> actions(size);
+    for (auto& action : actions) {
+        action.IsGet = RandomNumber<double>() < GetFrequency;
+        action.Key = ToString(RandomNumber(MaxKey));
+        action.Value = ToString(MinValue + RandomNumber(MaxValue - MinValue));
+    }
+    return actions;
+}
+
+TIntrusivePtr<TPausedClock> MakePausedClock() {
+    return new TPausedClock();
+}
+
+Y_UNIT_TEST_SUITE(LocalCacheTests) {
+
+    Y_UNIT_TEST(OnEmpty_WhenGet_ThenReturnedExpiredDefault) {
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {});
+
+        auto entry = cache->Get("1").GetValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL(entry.Value, "");
+        UNIT_ASSERT_VALUES_EQUAL(entry.IsExpired, true);
+    }
+
+    Y_UNIT_TEST(OnEmpty_WhenUpdate_ThenReturnedNew) {
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {});
+
+        cache->Update("1", "1").GetValueSync();
+
+        auto entry = cache->Get("1").GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL(entry.Value, "1");
+        UNIT_ASSERT_VALUES_EQUAL(entry.IsExpired, false);
+    }
+
+    Y_UNIT_TEST(OnExistingKey_WhenUpdate_ThenReturnedNew) {
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {});
+        cache->Update("1", "1");
+
+        cache->Update("1", "2").GetValueSync();
+
+        auto entry = cache->Get("1").GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL(entry.Value, "2");
+        UNIT_ASSERT_VALUES_EQUAL(entry.IsExpired, false);
+    }
+
+    Y_UNIT_TEST(OnExistingKey_WhenExpires_ThenReturnedOld) {
+        auto clock = MakePausedClock();
+        auto cache = MakeLocalCache<TString, TString>(clock, {.TTL = TDuration::Minutes(2)});
+        cache->Update("1", "1");
+
+        clock->Skip(TDuration::Minutes(2) + TDuration::Seconds(1));
+
+        auto entry = cache->Get("1").GetValueSync();
+        UNIT_ASSERT_VALUES_EQUAL(entry.Value, "1");
+        UNIT_ASSERT_VALUES_EQUAL(entry.IsExpired, true);
+    }
+
+    Y_UNIT_TEST(OnExistingKey_WhenGetResultExtracted_ThenItIsCopied) {
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {});
+        cache->Update("1", TString(128, '1'));
+
+        cache->Get("1").ExtractValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL(cache->Get("1").ExtractValueSync().Value, TString(128, '1'));
+        UNIT_ASSERT_VALUES_EQUAL(cache->Get("1").ExtractValueSync().Value, TString(128, '1'));
+    }
+
+    Y_UNIT_TEST(OnFull_WhenFatAdded_ThenSomeKeyIsEvicted) {
+        const size_t Overhead = TByteSize<TFat>()({}) / 10;
+
+        auto cache = MakeLocalCache<int, TFat>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(),
+            {.ByteCapacity = 4 * TByteSize<TFat>()({}) + Overhead});
+        cache->Update(1, {});
+        cache->Update(2, {});
+        cache->Update(3, {});
+        cache->Update(4, {});
+
+        cache->Update(5, {});
+
+        size_t evicted = 0;
+        for (auto x : {1, 2, 3, 4, 5}) {
+            if (cache->Get(x).GetValueSync().IsExpired) {
+                evicted += 1;
+            }
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(evicted, 1);
+    }
+
+    Y_UNIT_TEST(OnFull_WhenFatAdded_ThenKeyAndOverheadAreAccounted) {
+        const size_t Overhead = TByteSize<TFat>()({}) / 10;
+
+        auto cache = MakeLocalCache<TFat, TFat>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(),
+            {.ByteCapacity = 3 * 4 * TByteSize<TFat>()({}) + Overhead});
+        cache->Update(TFat{1}, {});
+        cache->Update(TFat{2}, {});
+        cache->Update(TFat{3}, {});
+        cache->Update(TFat{4}, {});
+
+        cache->Update(TFat{5}, {});
+
+        size_t evicted = 0;
+        for (auto x : {TFat{1}, TFat{2}, TFat{3}, TFat{4}, TFat{5}}) {
+            if (cache->Get(x).GetValueSync().IsExpired) {
+                evicted += 1;
+            }
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(evicted, 1);
+    }
+
+    Y_UNIT_TEST(WhenRandomlyAccessed_ThenDoesNotDie) {
+        constexpr size_t Iterations = 1024 * 1024;
+        SetRandomSeed(1);
+
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {.ByteCapacity = 64, .TTL = TDuration::MilliSeconds(1)});
+
+        for (auto&& a : GenerateRandomActions(Iterations)) {
+            if (a.IsGet) {
+                Y_DO_NOT_OPTIMIZE_AWAY(cache->Get(a.Key));
+            } else {
+                Y_DO_NOT_OPTIMIZE_AWAY(cache->Update(a.Key, std::move(a.Value)));
+            }
+        }
+    }
+
+    Y_UNIT_TEST(WhenConcurrentlyAccessed_ThenDoesNotDie) {
+        constexpr size_t Threads = 8;
+        constexpr size_t Iterations = Threads * 16 * 1024;
+        SetRandomSeed(1);
+
+        auto cache = MakeLocalCache<TString, TString>(
+            NMonotonic::CreateDefaultMonotonicTimeProvider(), {.ByteCapacity = 64, .TTL = TDuration::MilliSeconds(1)});
+
+        auto pool = CreateThreadPool(Threads);
+        for (auto&& a : GenerateRandomActions(Iterations)) {
+            Y_ENSURE(pool->AddFunc([cache, a = std::move(a)] {
+                if (a.IsGet) {
+                    Y_DO_NOT_OPTIMIZE_AWAY(cache->Get(a.Key));
+                } else {
+                    Y_DO_NOT_OPTIMIZE_AWAY(cache->Update(a.Key, std::move(a.Value)));
+                }
+            }));
+        }
+        pool->Stop();
+    }
+
+} // Y_UNIT_TEST_SUITE(LocalCacheTests)
diff --git a/yql/essentials/sql/v1/complete/name/cache/local/ut/ya.make b/yql/essentials/sql/v1/complete/name/cache/local/ut/ya.make
new file mode 100644
index 00000000000..85db0e61f88
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/local/ut/ya.make
@@ -0,0 +1,7 @@
+UNITTEST_FOR(yql/essentials/sql/v1/complete/name/cache/local)
+
+SRCS(
+    cache_ut.cpp
+)
+
+END()
diff --git a/yql/essentials/sql/v1/complete/name/cache/local/ya.make b/yql/essentials/sql/v1/complete/name/cache/local/ya.make
new file mode 100644
index 00000000000..433f40f4704
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/local/ya.make
@@ -0,0 +1,17 @@
+LIBRARY()
+
+SRCS(
+    cache.cpp
+)
+
+PEERDIR(
+    yql/essentials/sql/v1/complete/name/cache
+    library/cpp/cache
+    library/cpp/time_provider
+)
+
+END()
+
+RECURSE_FOR_TESTS(
+    ut
+)
diff --git a/yql/essentials/sql/v1/complete/name/cache/ut/ya.make b/yql/essentials/sql/v1/complete/name/cache/ut/ya.make
new file mode 100644
index 00000000000..c40d146dc28
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/ut/ya.make
@@ -0,0 +1,11 @@
+UNITTEST_FOR(yql/essentials/sql/v1/complete/name/cache)
+
+SRCS(
+    cached_ut.cpp
+)
+
+PEERDIR(
+    yql/essentials/sql/v1/complete/name/cache/local
+)
+
+END()
diff --git a/yql/essentials/sql/v1/complete/name/cache/ya.make b/yql/essentials/sql/v1/complete/name/cache/ya.make
new file mode 100644
index 00000000000..9d8c18417a4
--- /dev/null
+++ b/yql/essentials/sql/v1/complete/name/cache/ya.make
@@ -0,0 +1,21 @@
+LIBRARY()
+
+SRCS(
+    byte_size.cpp
+    cache.cpp
+    cached.cpp
+)
+
+PEERDIR(
+    library/cpp/threading/future
+)
+
+END()
+
+RECURSE(
+    local
+)
+
+RECURSE_FOR_TESTS(
+    ut
+)