Mitigate double notification of compiled pattern scenario (ydb-platform#10499) (ydb-platform#10530)

Author: abyss7

ydb/core/kqp/compile_service/kqp_compile_computation_pattern_service.cpp

@@ -62,7 +62,7 @@ class TKqpCompileComputationPatternService : public TActorBootstrapped<TKqpCompi
             timer.Reset();
 
             patternToCompile.Entry->Pattern->Compile({}, nullptr);
-            patternCache->NotifyPatternCompiled(patternToCompile.SerializedProgram, patternToCompile.Entry);
+            patternCache->NotifyPatternCompiled(patternToCompile.SerializedProgram);
             patternToCompile.Entry = nullptr;
 
             Counters->CompiledComputationPatterns->Inc();

ydb/library/yql/minikql/computation/mkql_computation_pattern_cache.cpp

@@ -69,15 +69,20 @@ class TComputationPatternLRUCache::TLRUPatternCacheImpl
         ClearIfNeeded();
     }
 
-    void NotifyPatternCompiled(const TString & serializedProgram, std::shared_ptr<TPatternCacheEntry>& entry) {
+    void NotifyPatternCompiled(const TString & serializedProgram) {
         auto it = SerializedProgramToPatternCacheHolder.find(serializedProgram);
         if (it == SerializedProgramToPatternCacheHolder.end()) {
             return;
         }
 
-        Y_ASSERT(entry->Pattern->IsCompiled());
+        const auto& entry = it->second.Entry;
+
+        Y_ENSURE(entry->Pattern->IsCompiled());
+
+        if (it->second.LinkedInCompiledPatternLRUList()) {
+            return;
+        }
 
-        Y_ASSERT(!it->second.LinkedInCompiledPatternLRUList());
         PromoteEntry(&it->second);
 
         ++CurrentCompiledPatternsSize;
@@ -290,9 +295,9 @@ void TComputationPatternLRUCache::EmplacePattern(const TString& serializedProgra
     }
 }
 
-void TComputationPatternLRUCache::NotifyPatternCompiled(const TString& serializedProgram, std::shared_ptr<TPatternCacheEntry> patternWithEnv) {
+void TComputationPatternLRUCache::NotifyPatternCompiled(const TString& serializedProgram) {
     std::lock_guard lock(Mutex);
-    Cache->NotifyPatternCompiled(serializedProgram, patternWithEnv);
+    Cache->NotifyPatternCompiled(serializedProgram);
 }
 
 size_t TComputationPatternLRUCache::GetSize() const {

ydb/library/yql/minikql/computation/mkql_computation_pattern_cache.h

@@ -130,7 +130,7 @@ class TComputationPatternLRUCache {
 
     void EmplacePattern(const TString& serializedProgram, std::shared_ptr<TPatternCacheEntry> patternWithEnv);
 
-    void NotifyPatternCompiled(const TString& serializedProgram, std::shared_ptr<TPatternCacheEntry> patternWithEnv);
+    void NotifyPatternCompiled(const TString& serializedProgram);
 
     size_t GetSize() const;
 

ydb/library/yql/minikql/computation/mkql_computation_pattern_cache_ut.cpp

@@ -630,6 +630,42 @@ Y_UNIT_TEST_SUITE(ComputationPatternCache) {
         }
     }
 
+    Y_UNIT_TEST(DoubleNotifyPatternCompiled) {
+        class TMockComputationPattern final : public IComputationPattern {
+        public:
+            explicit TMockComputationPattern(size_t codeSize) : Size_(codeSize) {}
+
+            void Compile(TString, IStatsRegistry*) override { Compiled_ = true; }
+            bool IsCompiled() const override { return Compiled_; }
+            size_t CompiledCodeSize() const override { return Size_; }
+            void RemoveCompiledCode() override { Compiled_ = false; }
+            THolder<IComputationGraph> Clone(const TComputationOptsFull&) override { return {}; }
+            bool GetSuitableForCache() const override { return true; }
+
+        private:
+            const size_t Size_;
+            bool Compiled_ = false;
+        };
+
+        const TString key = "program";
+        const ui32 cacheSize = 2;
+        TComputationPatternLRUCache cache({cacheSize, cacheSize});
+
+        auto entry = std::make_shared<TPatternCacheEntry>();
+        entry->Pattern = MakeIntrusive<TMockComputationPattern>(1u);
+        cache.EmplacePattern(key, entry);
+
+        for (ui32 i = 0; i < cacheSize + 1; ++i) {
+            entry->Pattern->Compile("", nullptr);
+            cache.NotifyPatternCompiled(key);
+        }
+
+        entry = std::make_shared<TPatternCacheEntry>();
+        entry->Pattern = MakeIntrusive<TMockComputationPattern>(cacheSize + 1);
+        entry->Pattern->Compile("", nullptr);
+        cache.EmplacePattern(key, entry);
+    }
+
     Y_UNIT_TEST(AddPerf) {
         TTimer t("all: ");
         TScopedAlloc alloc(__LOCATION__);