fixed all tests

Author: lll-phill-lll

ydb/library/yql/dq/actors/spilling/spilling_file.h

@@ -17,7 +17,7 @@ struct TFileSpillingServiceConfig {
     ui64 MaxFilePartSize = 0;
 
     ui32 IoThreadPoolWorkersCount = 2;
-    ui32 IoThreadPoolQueueSize = 1000;
+    ui32 IoThreadPoolQueueSize = 100000;
     bool CleanupOnShutdown = false;
 };
 

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

@@ -596,6 +596,10 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
     }
 
     EFetchResult FetchValues(TComputationContext& ctx, NUdf::TUnboxedValue*const* output) {
+        if (!IsLogPrinted) {
+            std::cerr << "[MISHA] new join\n";
+            IsLogPrinted = true;
+        }
         while (true) {
             switch(GetMode()) {
                 case EOperatingMode::InMemory: {
@@ -622,6 +626,8 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
     }
 
 private:
+    bool IsLogPrinted = false;
+
     EOperatingMode GetMode() const {
         return Mode;
     }
@@ -800,6 +806,7 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
             }
 
         }
+        std::cerr << "[MISHA] join finished\n";
 
         return EFetchResult::Finish;
     }
@@ -832,7 +839,7 @@ class TGraceJoinSpillingSupportState : public TComputationValue<TGraceJoinSpilli
 void DoCalculateWithSpilling(TComputationContext& ctx) {
     UpdateSpilling();
 
-    if (!HasMemoryForProcessing()) {
+    if (!HasMemoryForProcessing() && !IsSpillingFinalized) {
         bool isWaitingForReduce = TryToReduceMemoryAndWait();
         if (isWaitingForReduce) return;
     }
@@ -906,6 +913,7 @@ EFetchResult ProcessSpilledData(TComputationContext&, NUdf::TUnboxedValue*const*
 
                 NextBucketToJoin++;
             } else {
+                std::cerr << "[MISHA] joining\n";
                 *PartialJoinCompleted = true;
                 LeftPacker->StartTime = std::chrono::system_clock::now();
                 RightPacker->StartTime = std::chrono::system_clock::now();

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

@@ -1172,7 +1172,10 @@ void TTable::UpdateSpilling() {
 
 bool TTable::IsSpillingFinished() const {
     for (ui64 i = 0; i < NumberOfBuckets; ++i) {
-        if (TableBucketsSpillers[i].IsProcessingSpilling()) return false;
+        if (TableBucketsSpillers[i].IsProcessingSpilling()) {
+            std::cerr << std::format("[MISHA] bucket {} still spilling : {}\n", i, TableBucketsSpillers[i].GetStateName());
+            return false;
+        }
     }
     return true;
 }