YQL-18468 fix wrong gracejoin answer with spilling on q12 tpch

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,13 +632,15 @@ 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) {
+
+        std::cerr << std::format("[MISHA] switching mode {}->{}\n", (int)Mode, (int)mode);
         switch(mode) {
             case EOperatingMode::InMemory: {
                 MKQL_ENSURE(false, "Internal logic error");
@@ -772,7 +775,7 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
             }
             if (isYield) return EFetchResult::Yield;
 
-            if (!*HaveMoreRightRows && !*PartialJoinCompleted && LeftPacker->TuplesBatchPacked >= LeftPacker->BatchSize ) {
+            /*if (!*HaveMoreRightRows && !*PartialJoinCompleted && LeftPacker->TuplesBatchPacked >= LeftPacker->BatchSize ) {
                 *PartialJoinCompleted = true;
                 JoinedTablePtr->Join(*LeftPacker->TablePtr, *RightPacker->TablePtr, JoinKind, *HaveMoreLeftRows, *HaveMoreRightRows);
                 JoinedTablePtr->ResetIterator();
@@ -783,7 +786,7 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
                 JoinedTablePtr->Join(*LeftPacker->TablePtr, *RightPacker->TablePtr, JoinKind, *HaveMoreLeftRows, *HaveMoreRightRows);
                 JoinedTablePtr->ResetIterator();
 
-            }
+            }*/
 
             if (!*HaveMoreRightRows && !*HaveMoreLeftRows && !*PartialJoinCompleted) {
                 *PartialJoinCompleted = true;
@@ -855,10 +858,12 @@ EFetchResult ProcessSpilledData(TComputationContext&, NUdf::TUnboxedValue*const*
         if (HasRunningAsyncOperation()) return EFetchResult::Yield;
 
         if (!LeftPacker->TablePtr->IsBucketInMemory(NextBucketToJoin)) {
+            std::cerr << std::format("[MISHA] loading bucket {}\n", NextBucketToJoin);
             LeftPacker->TablePtr->StartLoadingBucket(NextBucketToJoin);
         }
 
         if (!RightPacker->TablePtr->IsBucketInMemory(NextBucketToJoin)) {
+            std::cerr << std::format("[MISHA] loading bucket {}\n", NextBucketToJoin);
             RightPacker->TablePtr->StartLoadingBucket(NextBucketToJoin);
         } 
 

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>
 
@@ -263,6 +264,7 @@ void ResizeHashTable(KeysHashTable &t, ui64 newSlots){
 // Joins two tables and returns join result in joined table. Tuples of joined table could be received by
 // joined table iterator
 void TTable::Join( TTable & t1, TTable & t2, EJoinKind joinKind, bool hasMoreLeftTuples, bool hasMoreRightTuples, ui32 fromBucket, ui32 toBucket ) {
+    std::cerr << std::format("[MISHA] joining buckets {}->{}\n", fromBucket, toBucket);
     if ( hasMoreLeftTuples )
         LeftTableBatch_ = true;
 
@@ -297,13 +299,17 @@ void TTable::Join( TTable & t1, TTable & t2, EJoinKind joinKind, bool hasMoreLef
 
     for (ui64 bucket = fromBucket; bucket < toBucket; bucket++) {
 
+        // std::cerr << std::format("[MISHA] bucket: {}", bucket);
+
         joinResults.clear();
         TTableBucket * bucket1 = &JoinTable1->TableBuckets[bucket];
         TTableBucket * bucket2 = &JoinTable2->TableBuckets[bucket];
 
         ui64 tuplesNum1 = JoinTable1->TableBucketsStats[bucket].TuplesNum;
         ui64 tuplesNum2 = JoinTable2->TableBucketsStats[bucket].TuplesNum;
 
+        std::cerr << std::format("[MISHA] bucket: {}, tuplesNum1: {}, tuplesNum2: {}, KIVSize1: {}, KIVSize2: {}\n", bucket, tuplesNum1, tuplesNum2, bucket1->KeyIntVals.size(), bucket2->KeyIntVals.size());
+
         ui64 headerSize1 = JoinTable1->HeaderSize;
         ui64 headerSize2 = JoinTable2->HeaderSize;
         ui64 nullsSize1 = JoinTable1->NullsBitmapSize_;
@@ -1154,8 +1160,8 @@ bool TTable::TryToReduceMemoryAndWait() {
         }
     }
 
-    if (largestBucketSize) return false;
-
+    if (!largestBucketSize) return false;
+    std::cerr << std::format("[MISHA] spilling bucket {} of size {}\n", largestBucketIndex, largestBucketSize);
     TableBucketsSpillers[largestBucketIndex].SpillBucket(std::move(TableBuckets[largestBucketIndex]));
     TableBuckets[largestBucketIndex] = TTableBucket{};
 

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;