Merge 170b5d9 into 89dfce5

Author: lll-phill-lll

ydb/library/yql/minikql/comp_nodes/mkql_wide_combine.cpp

@@ -1,6 +1,7 @@
 #include "mkql_wide_combine.h"
 #include "mkql_rh_hash.h"
 
+#include <format>
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_codegen.h>  // Y_IGNORE
 #include <ydb/library/yql/minikql/computation/mkql_llvm_base.h>  // Y_IGNORE
 #include <ydb/library/yql/minikql/computation/mkql_computation_node.h>
@@ -550,17 +551,19 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
     }
 
     bool CheckMemoryAndSwitchToSpilling() {
+        std::cerr << std::format("MISHA: {} {} {}\n", AllowSpilling, (bool)Ctx.SpillerFactory, IsSwitchToSpillingModeCondition());
         if (AllowSpilling && Ctx.SpillerFactory && IsSwitchToSpillingModeCondition()) {
-            const auto used = TlsAllocState->GetUsed();
-            const auto limit = TlsAllocState->GetLimit();
+            std::cerr << "MISHA spilling enabled" << std::endl;
+            // const auto used = TlsAllocState->GetUsed();
+            // const auto limit = TlsAllocState->GetLimit();
 
-            YQL_LOG(INFO) << "yellow zone reached " << (used*100/limit) << "%=" << used << "/" << limit;
+            // YQL_LOG(INFO) << "yellow zone reached " << (used*100/limit) << "%=" << used << "/" << limit;
             YQL_LOG(INFO) << "switching Memory mode to Spilling";
 
             SwitchMode(EOperatingMode::Spilling);
             return true;
         }
-
+        std::cerr << "MISHA spilling not enabled" << std::endl;
         return false;
     }
 
@@ -722,9 +725,7 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
     }
 
     bool IsSwitchToSpillingModeCondition() const {
-        return false;
-        // TODO: YQL-18033
-        // return !HasMemoryForProcessing();
+        return !HasMemoryForProcessing();
     }
 
 public:

ydb/library/yql/minikql/comp_nodes/ut/mkql_computation_node_ut.h

@@ -48,6 +48,7 @@
     void N(NUnitTest::TTestContext&)
 
 #define Y_UNIT_TEST_LLVM(N) Y_UNIT_TEST_TWIN(N, LLVM)
+#define Y_UNIT_TEST_LLVM_SPILLING(N) Y_UNIT_TEST_QUAD(N, LLVM, SPILLING)
 
 #define Y_UNIT_TEST_QUAD(N, OPT1, OPT2)                                                                                              \
     template<bool OPT1, bool OPT2> void N(NUnitTest::TTestContext&);                                                                 \
@@ -79,7 +80,7 @@ struct TUdfModuleInfo {
     NUdf::TUniquePtr<NUdf::IUdfModule> Module;
 };
 
-template<bool UseLLVM>
+template<bool UseLLVM, bool EnableSpilling = false>
 struct TSetup {
     explicit TSetup(TComputationNodeFactory nodeFactory = GetTestFactory(), TVector<TUdfModuleInfo>&& modules = {})
         : Alloc(__LOCATION__)
@@ -96,6 +97,8 @@ struct TSetup {
             FunctionRegistry = mutableRegistry;
         }
 
+        Alloc.Ref().ForcefullySetMemoryYellowZone(EnableSpilling);
+
         RandomProvider = CreateDeterministicRandomProvider(1);
         TimeProvider = CreateDeterministicTimeProvider(10000000);
 

ydb/library/yql/minikql/comp_nodes/ut/mkql_test_factory.cpp

@@ -2,6 +2,7 @@
 
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_impl.h>
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_holders.h>
+#include <ydb/library/yql/minikql/computation/mock_spiller_factory_ut.h>
 #include <ydb/library/yql/minikql/mkql_node_cast.h>
 #include <ydb/library/yql/minikql/mkql_string_util.h>
 
@@ -51,6 +52,7 @@ class TTestStreamWrapper: public TMutableComputationNode<TTestStreamWrapper> {
     }
 
     NUdf::TUnboxedValuePod DoCalculate(TComputationContext& ctx) const {
+        ctx.SpillerFactory = std::make_shared<TMockSpillerFactory>();
         return ctx.HolderFactory.Create<TStreamValue>(Count);
     }
 
@@ -70,7 +72,9 @@ class TTestYieldStreamWrapper: public TMutableComputationNode<TTestYieldStreamWr
 
         TStreamValue(TMemoryUsageInfo* memInfo, TComputationContext& compCtx)
             : TBase(memInfo)
-            , CompCtx(compCtx) {}
+            , CompCtx(compCtx) {
+                CompCtx.SpillerFactory = std::make_shared<TMockSpillerFactory>();
+            }
 
     private:
         NUdf::EFetchStatus Fetch(NUdf::TUnboxedValue& result) override {
@@ -102,6 +106,7 @@ class TTestYieldStreamWrapper: public TMutableComputationNode<TTestYieldStreamWr
         : TBaseComputation(mutables) {}
 
     NUdf::TUnboxedValuePod DoCalculate(TComputationContext& ctx) const {
+        
         return ctx.HolderFactory.Create<TStreamValue>(ctx);
     }