added debug log + spilling

Author: lll-phill-lll

ydb/library/yql/dq/actors/compute/dq_compute_memory_quota.h

@@ -1,6 +1,7 @@
 
 #pragma once
 
+#include <format>
 #include <util/system/mem_info.h>
 #include <ydb/library/services/services.pb.h>
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
@@ -60,15 +61,17 @@ namespace NYql::NDq {
         // Most likely this callback will be removed after KIKIMR-21481.
         void TrySetIncreaseMemoryLimitCallbackWithRSSControl(NKikimr::NMiniKQL::TScopedAlloc* alloc) {
             if (!CanAllocateExtraMemory) return;
-            const ui64 limitRSS = std::numeric_limits<ui64>::max();
+            const ui64 limitRSS = 3_GB;
             const ui64 criticalRSSValue = limitRSS / 100 * 80;
 
             alloc->Ref().SetIncreaseMemoryLimitCallback([this, alloc](ui64 limit, ui64 required) {
                 RequestExtraMemory(required - limit, alloc);
 
                 ui64 currentRSS = NMemInfo::GetMemInfo().RSS;
+                std::cerr << std::format("[MISHA]: limit: {}MB required: {}MB, RSS: {}MB\n", limit / 1024 / 1024, required / 1024 / 1024, currentRSS / 1024 / 1024);
                 if (currentRSS > criticalRSSValue) {
                     alloc->SetMaximumLimitValueReached(true);
+                    std::cerr << std::format("[MISHA]: MAximum limit set up\n");
                 }
             });
         }

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>
@@ -366,7 +367,13 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
     ~TSpillingSupportState() {
     }
 
+    bool WasLog = false;
+
     EFetchResult DoCalculate(TComputationContext& ctx, NUdf::TUnboxedValue*const* output) {
+        if (!WasLog) {
+            std::cerr << "[MISHA] WIDE COMBINE\n";
+            WasLog = true;
+        }
         while (true) {
             switch(GetMode()) {
                 case EOperatingMode::InMemory: {
@@ -514,8 +521,10 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
 
         if (!HasMemoryForProcessing()) {
             bool isWaitingForReduce = TryToReduceMemory();
+            std::cerr << std::format("[MISHA] PROCESSING. is waiting: {}\n", isWaitingForReduce);
             if (isWaitingForReduce) return;
         }
+        std::cerr << "[MISHA] PROCESSING\n";
 
         if (BufferForUsedInputItems.size()) {
             auto& bucket = SpilledBuckets[BufferForUsedInputItemsBucketId];
@@ -680,6 +689,7 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
                 break;
             }
             case EOperatingMode::Spilling: {
+                std::cerr << "[MISHA] spilling enabled\n";
                 MKQL_ENSURE(EOperatingMode::InMemory == Mode, "Internal logic error");
                 SpilledBuckets.resize(SpilledBucketCount);
                 auto spiller = ctx.SpillerFactory->CreateSpiller();
@@ -706,9 +716,9 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
     }
 
     bool IsSwitchToSpillingModeCondition() const {
-        return false;
+        // return false;
         // TODO: YQL-18033
-        // return !HasMemoryForProcessing();
+        return !HasMemoryForProcessing();
     }
 
 private:

ydb/library/yql/providers/dq/worker_manager/local_worker_manager.h

@@ -30,7 +30,7 @@ namespace NYql::NDqs {
         NDq::NTaskRunnerActor::ITaskRunnerActorFactory::TPtr TaskRunnerActorFactory;
         THashMap<TString, TString> ClusterNamesMapping;
 
-        ui64 MkqlInitialMemoryLimit = 8_GB;
+        ui64 MkqlInitialMemoryLimit = 300_MB;
         ui64 MkqlTotalMemoryLimit = 0;
         ui64 MkqlMinAllocSize = 30_MB;
         ui64 MkqlProgramHardMemoryLimit = 0;