start using spilling in YDB (#5496)

Author: gridnevvvit

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp

@@ -13,13 +13,15 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
         , std::shared_ptr<IKqpNodeState> state
         , ui64 txId
         , ui64 taskId
-        , ui64 limit)
+        , ui64 limit
+        , ui64 reasonableSpillingTreshold)
     : NYql::NDq::TGuaranteeQuotaManager(limit, limit)
     , ResourceManager(std::move(resourceManager))
     , MemoryPool(memoryPool)
     , State(std::move(state))
     , TxId(txId)
     , TaskId(taskId)
+    , ReasonableSpillingTreshold(reasonableSpillingTreshold)
     {
     }
 
@@ -42,6 +44,8 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
             return false;
         }
 
+        TotalQueryAllocationsSize = result.TotalAllocatedQueryMemory;
+
         return true;
     }
 
@@ -51,6 +55,10 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
         );
     }
 
+    bool IsReasonableToUseSpilling() const override {
+        return TotalQueryAllocationsSize >= ReasonableSpillingTreshold;
+    }
+
     void TerminateHandler(bool success, const NYql::TIssues& issues) {
         AFL_DEBUG(NKikimrServices::KQP_COMPUTE)
             ("problem", "finish_compute_actor")
@@ -64,6 +72,8 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
     ui64 TxId;
     ui64 TaskId;
     bool Success = true;
+    ui64 TotalQueryAllocationsSize = 0;
+    ui64 ReasonableSpillingTreshold = 0;
 };
 
 class TKqpCaFactory : public IKqpNodeComputeActorFactory {
@@ -123,7 +133,8 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
             std::move(state),
             txId,
             dqTask->GetId(),
-            limit);
+            limit,
+            Config.GetReasonableSpillingTreshold());
 
         auto runtimeSettings = settings;
         NYql::NDq::IMemoryQuotaManager::TWeakPtr memoryQuotaManager = memoryLimits.MemoryQuotaManager;

ydb/core/kqp/counters/kqp_counters.cpp

@@ -726,7 +726,7 @@ void TKqpCounters::UpdateTxCounters(const TKqpTransactionInfo& txInfo,
 }
 
 TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, const TActorContext* ctx)
-    : NYql::NDq::TSpillingCounters(counters)
+    : NYql::NDq::TSpillingCounters(GetServiceCounters(counters, "kqp"))
     , AllocCounters(counters, "kqp")
 {
     Counters = counters;

ydb/core/kqp/rm_service/kqp_rm_service.cpp

@@ -308,6 +308,7 @@ class TKqpResourceManager : public IKqpResourceManager {
 
             auto& tx = txBucket.Txs[txId];
             ui64 txTotalRequestedMemory = tx.TxScanQueryMemory + resources.Memory;
+            result.TotalAllocatedQueryMemory = txTotalRequestedMemory;
             if (txTotalRequestedMemory > queryMemoryLimit) {
                 TStringBuilder reason;
                 reason << "TxId: " << txId << ", taskId: " << taskId << ". Query memory limit exceeded: "

ydb/core/kqp/rm_service/kqp_rm_service.h

@@ -51,6 +51,7 @@ struct TKqpRMAllocateResult {
     bool Success = true;
     NKikimrKqp::TEvStartKqpTasksResponse::ENotStartedTaskReason Status = NKikimrKqp::TEvStartKqpTasksResponse::INTERNAL_ERROR;
     TString FailReason;
+    ui64 TotalAllocatedQueryMemory = 0;
 
     NKikimrKqp::TEvStartKqpTasksResponse::ENotStartedTaskReason GetStatus() const {
         return Status;

ydb/core/protos/table_service_config.proto

@@ -42,6 +42,7 @@ message TTableServiceConfig {
         optional TInfoExchangerSettings InfoExchangerSettings = 19;
         optional uint64 KqpPatternCachePatternAccessTimesBeforeTryToCompile = 20 [default = 5];
         optional uint64 KqpPatternCacheCompiledCapacityBytes = 21 [default = 104857600]; // 100 MiB
+        optional uint64 ReasonableSpillingTreshold = 22 [default = 104857600]; // 100 MiB
     }
 
     message TSpillingServiceConfig {

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

@@ -307,6 +307,10 @@ struct TGuaranteeQuotaManager : public IMemoryQuotaManager {
         return true;
     }
 
+    bool IsReasonableToUseSpilling() const override {
+        return false;
+    }
+
     void FreeQuota(ui64 memorySize) override {
         Y_ABORT_UNLESS(Quota >= memorySize);
         Quota -= memorySize;

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

@@ -67,6 +67,7 @@ struct IMemoryQuotaManager {
     virtual void FreeQuota(ui64 memorySize) = 0;
     virtual ui64 GetCurrentQuota() const = 0;
     virtual ui64 GetMaxMemorySize() const = 0;
+    virtual bool IsReasonableToUseSpilling() const = 0;
 };
 
 // Source/transform.

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

@@ -65,7 +65,7 @@ namespace NYql::NDq {
 
             alloc->Ref().SetIncreaseMemoryLimitCallback([this, alloc](ui64 limit, ui64 required) {
                 RequestExtraMemory(required - limit, alloc);
-                
+
                 ui64 currentRSS = NMemInfo::GetMemInfo().RSS;
                 if (currentRSS > criticalRSSValue) {
                     alloc->SetMaximumLimitValueReached(true);
@@ -150,6 +150,12 @@ namespace NYql::NDq {
                 //                << ", requested: " << memory << ", host: " << HostName();
             }
 
+            if (MemoryLimits.MemoryQuotaManager->IsReasonableToUseSpilling()) {
+                alloc->SetMaximumLimitValueReached(true);
+            } else {
+                alloc->SetMaximumLimitValueReached(false);
+            }
+
             if (Y_UNLIKELY(ProfileStats)) {
                 ProfileStats->MkqlExtraMemoryBytes += memory;
                 ProfileStats->MkqlExtraMemoryRequests++;

ydb/library/yql/providers/dq/actors/worker_actor.cpp

@@ -81,6 +81,10 @@ class TDummyMemoryQuotaManager: public IMemoryQuotaManager {
     ui64 GetMaxMemorySize() const override {
         return std::numeric_limits<ui64>::max();
     }
+
+    bool IsReasonableToUseSpilling() const override {
+        return false;
+    }
 };
 
 class TDqWorker: public TRichActor<TDqWorker>