Merge 5302382 into c808caf

Author: lll-phill-lll

ydb/library/yql/minikql/aligned_page_pool.cpp

@@ -504,7 +504,7 @@ void TAlignedPagePoolImpl<T>::Free(void* ptr, size_t size) noexcept {
 template<typename T>
 void TAlignedPagePoolImpl<T>::UpdateMemoryYellowZone() {
     if (Limit == 0) return;
-    if (IncreaseMemoryLimitCallback && !IsMaximumLimitValueReached) return;
+    // if (IncreaseMemoryLimitCallback && !IsMaximumLimitValueReached) return;
 
     ui8 usedMemoryPercent = 100 * GetUsed() / Limit;
     if (usedMemoryPercent >= EnableMemoryYellowZoneThreshold) {

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

@@ -1,6 +1,7 @@
 #include "mkql_grace_join.h"
 #include "mkql_grace_join_imp.h"
 
+#include <format>
 #include <ydb/library/yql/public/udf/udf_data_type.h>
 #include <ydb/library/yql/public/udf/udf_value.h>
 #include <ydb/library/yql/public/decimal/yql_decimal_serialize.h>
@@ -631,9 +632,9 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
     }
 
     bool IsSwitchToSpillingModeCondition() const {
-        return false;
+        // return false;
         // TODO: YQL-18033
-        // return !HasMemoryForProcessing();
+        return !HasMemoryForProcessing();
     }
 
     void SwitchMode(EOperatingMode mode, TComputationContext& ctx) {
@@ -804,7 +805,10 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
     }
 
     bool TryToReduceMemoryAndWait() {
+        std::cerr << std::format("[MISHA][LEFT] Trying to reduce memory {}MB/{}MB\n", TlsAllocState->GetAllocated() / 1024 / 1024, TlsAllocState->GetLimit() / 1024 / 1024);
         bool isWaitingLeftForReduce = LeftPacker->TablePtr->TryToReduceMemoryAndWait();
+        if (isWaitingLeftForReduce) return true;
+        std::cerr << std::format("[MISHA][RIGHT] Trying to reduce memory {}MB/{}MB\n", TlsAllocState->GetAllocated() / 1024 / 1024, TlsAllocState->GetLimit() / 1024 / 1024);
         bool isWaitingRightForReduce = RightPacker->TablePtr->TryToReduceMemoryAndWait();
 
         return isWaitingLeftForReduce || isWaitingRightForReduce;
@@ -862,6 +866,8 @@ EFetchResult ProcessSpilledData(TComputationContext&, NUdf::TUnboxedValue*const*
         UpdateSpilling();
         if (IsRestoringSpilledBuckets()) return EFetchResult::Yield;
 
+        
+
         if (LeftPacker->TablePtr->IsSpilledBucketWaitingForExtraction(NextBucketToJoin)) {
             LeftPacker->TablePtr->PrepareBucket(NextBucketToJoin);
         }
@@ -885,6 +891,8 @@ EFetchResult ProcessSpilledData(TComputationContext&, NUdf::TUnboxedValue*const*
                     return EFetchResult::One;
                 }
 
+                std::cerr << std::format("[MISHA] Bucket {} joined {}MB/{}MB\n", NextBucketToJoin, TlsAllocState->GetAllocated() / 1024 / 1024, TlsAllocState->GetLimit() / 1024 / 1024);
+
                 LeftPacker->TuplesBatchPacked = 0;
                 LeftPacker->TablePtr->ClearBucket(NextBucketToJoin); // Clear content of returned bucket
 
@@ -895,6 +903,8 @@ EFetchResult ProcessSpilledData(TComputationContext&, NUdf::TUnboxedValue*const*
                 JoinedTablePtr->ResetIterator();
                 *PartialJoinCompleted = false;
 
+                std::cerr << std::format("[MISHA] Bucket {} cleared {}MB/{}MB\n", NextBucketToJoin, TlsAllocState->GetAllocated() / 1024 / 1024, TlsAllocState->GetLimit() / 1024 / 1024);
+
                 NextBucketToJoin++;
             } else {
                 *PartialJoinCompleted = true;

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

@@ -1,5 +1,6 @@
 #include "mkql_grace_join_imp.h"
 
+#include <format>
 #include <ydb/library/yql/public/udf/udf_data_type.h>
 #include <ydb/library/yql/utils/log/log.h>
 
@@ -1113,12 +1114,19 @@ void TTable::Clear() {
 
 void TTable::ClearBucket(ui64 bucket) {
     TTableBucket & tb = TableBuckets[bucket];
-    tb.KeyIntVals.clear();
+    std::vector<ui64, TMKQLAllocator<ui64>>().swap(tb.KeyIntVals);
+    std::vector<ui64, TMKQLAllocator<ui64>>().swap(tb.DataIntVals);
+    std::vector<ui32, TMKQLAllocator<ui32>>().swap(tb.StringsOffsets);
+    std::vector<char, TMKQLAllocator<char>>().swap(tb.StringsValues);
+    std::vector<char, TMKQLAllocator<char>>().swap(tb.InterfaceValues);
+    std::vector<ui32, TMKQLAllocator<ui32>>().swap(tb.InterfaceOffsets);
+
+    /* tb.KeyIntVals.clear();
     tb.DataIntVals.clear();
     tb.StringsOffsets.clear();
     tb.StringsValues.clear();
     tb.InterfaceValues.clear();
-    tb.InterfaceOffsets.clear();
+    tb.InterfaceOffsets.clear();*/
     tb.JoinIds.clear();
     tb.RightIds.clear();
 
@@ -1147,8 +1155,10 @@ bool TTable::TryToReduceMemoryAndWait() {
     i32 largestBucketIndex = 0;
     ui64 largestBucketSize = 0;
     for (ui32 bucket = 0; bucket < NumberOfBuckets; ++bucket) {
-        if (TableBucketsSpillers[bucket].IsProcessingSpilling()) return true;
-
+        if (TableBucketsSpillers[bucket].IsProcessingSpilling())  {
+            std::cerr << std::format("[MISHA] NOT spilling because of bucket {}\n", bucket);
+            return true;
+        }
         ui64 bucketSize = GetSizeOfBucket(bucket);
         if (bucketSize > largestBucketSize) {
             largestBucketSize = bucketSize;
@@ -1157,6 +1167,8 @@ bool TTable::TryToReduceMemoryAndWait() {
     }
 
     if (!largestBucketSize) return false;
+    TotalSpilled += largestBucketSize;
+    std::cerr << std::format("[MISHA][{}MB] spilling {} of size {}KB\n", TotalSpilled / 1024 / 1024, largestBucketIndex, largestBucketSize / 1024);
     TableBucketsSpillers[largestBucketIndex].SpillBucket(std::move(TableBuckets[largestBucketIndex]));
     TableBuckets[largestBucketIndex] = TTableBucket{};
 

ydb/library/yql/minikql/comp_nodes/mkql_grace_join_imp.h

@@ -172,6 +172,7 @@ class TTableBucketSpiller {
 
 // Class which represents single table data stored in buckets
 class TTable {
+    ui64 TotalSpilled = 0;
     ui64 NumberOfKeyIntColumns = 0; // Key int columns always first and padded to sizeof(ui64).
     ui64 NumberOfKeyStringColumns = 0; // String key columns go after key int columns
     ui64 NumberOfKeyIColumns = 0; // Number of interface - provided key columns

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 = 50_MB;
         ui64 MkqlTotalMemoryLimit = 0;
         ui64 MkqlMinAllocSize = 30_MB;
         ui64 MkqlProgramHardMemoryLimit = 0;