Merge 71c80ed into 63653ea

Author: gridnevvvit

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp

@@ -63,6 +63,10 @@ struct TMemoryQuotaManager : public NYql::NDq::TGuaranteeQuotaManager {
         return TotalQueryAllocationsSize >= ReasonableSpillingTreshold;
     }
 
+    TString MemoryConsumptionDetails() const override {
+        return ResourceManager->GetTxResourcesUsageDebugInfo(TxId);
+    }
+
     void TerminateHandler(bool success, const NYql::TIssues& issues) {
         AFL_DEBUG(NKikimrServices::KQP_COMPUTE)
             ("problem", "finish_compute_actor")

ydb/core/kqp/rm_service/kqp_rm_service.cpp

@@ -107,6 +107,15 @@ struct TTxState {
     ui32 TxExecutionUnits = 0;
     TInstant CreatedAt;
 
+    TString ToString() const {
+        return TStringBuilder() << "TxResourcesInfo{ "
+            << "Memory initially granted resources: " << TxExternalDataQueryMemory
+            << ", extra allocations " << TxScanQueryMemory
+            << ", execution units: " << TxExecutionUnits
+            << ", started at: " << CreatedAt
+            << " }";
+    }
+
     TTaskState& Allocated(ui64 taskId, TInstant now, const TKqpResourcesRequest& resources, bool memoryAsExternal = false) {
         ui64 externalMemory = resources.ExternalMemory;
         ui64 resourceBrokerMemory = 0;
@@ -564,6 +573,19 @@ class TKqpResourceManager : public IKqpResourceManager {
         ActorSystem->Send(SelfId, new TEvPrivate::TEvSchedulePublishResources);
     }
 
+    TString GetTxResourcesUsageDebugInfo(ui64 txId) override {
+        auto& txBucket = TxBucket(txId);
+        with_lock(txBucket.Lock) {
+            auto it = txBucket.Txs.find(txId);
+            if (it == txBucket.Txs.end()) {
+                return "<empty info>";
+            }
+
+            return it->second.ToString();
+        }
+    }
+
+
     void UpdatePatternCache(ui64 maxSizeBytes, ui64 maxCompiledSizeBytes, ui64 patternAccessTimesBeforeTryToCompile) {
         if (maxSizeBytes == 0) {
             PatternCache.reset();

ydb/core/kqp/rm_service/kqp_rm_service.h

@@ -94,6 +94,7 @@ class IKqpResourceManager : private TNonCopyable {
 
     virtual void FreeResources(ui64 txId, ui64 taskId, const TKqpResourcesRequest& resources) = 0;
     virtual void FreeResources(ui64 txId, ui64 taskId) = 0;
+    virtual TString GetTxResourcesUsageDebugInfo(ui64 txId) = 0;
 
     virtual void NotifyExternalResourcesAllocated(ui64 txId, ui64 taskId, const TKqpResourcesRequest& resources) = 0;
 

ydb/core/kqp/ut/query/kqp_limits_ut.cpp

@@ -128,6 +128,27 @@ Y_UNIT_TEST_SUITE(KqpLimits) {
         UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::BAD_REQUEST);
     }
 
+    Y_UNIT_TEST(ComputeActorMemoryAllocationFailure) {
+        auto app = NKikimrConfig::TAppConfig();
+        app.MutableTableServiceConfig()->MutableResourceManager()->SetMkqlLightProgramMemoryLimit(10);
+        app.MutableTableServiceConfig()->MutableResourceManager()->SetQueryMemoryLimit(2000);
+
+        TKikimrRunner kikimr(app);
+        CreateLargeTable(kikimr, 0, 0, 0);
+
+        kikimr.GetTestServer().GetRuntime()->SetLogPriority(NKikimrServices::KQP_SLOW_LOG, NActors::NLog::PRI_ERROR);
+
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto result = session.ExecuteDataQuery(Q1_(R"(
+            SELECT * FROM `/Root/LargeTable`;
+        )"), TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+        result.GetIssues().PrintTo(Cerr);
+
+        UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::OVERLOADED);
+    }
+
     Y_UNIT_TEST(DatashardProgramSize) {
         auto app = NKikimrConfig::TAppConfig();
         app.MutableTableServiceConfig()->MutableResourceManager()->SetMkqlLightProgramMemoryLimit(1'000'000'000);

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

@@ -331,6 +331,10 @@ struct TGuaranteeQuotaManager : public IMemoryQuotaManager {
         return MaxMemorySize;
     };
 
+    TString MemoryConsumptionDetails() const override {
+        return TString();
+    }
+
     virtual bool AllocateExtraQuota(ui64) {
         return false;
     }

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

@@ -68,6 +68,7 @@ struct IMemoryQuotaManager {
     virtual ui64 GetCurrentQuota() const = 0;
     virtual ui64 GetMaxMemorySize() const = 0;
     virtual bool IsReasonableToUseSpilling() const = 0;
+    virtual TString MemoryConsumptionDetails() const = 0;
 };
 
 // Source/transform.

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

@@ -147,10 +147,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
 
             static_cast<TDerived*>(this)->DoBootstrap();
         } catch (const NKikimr::TMemoryLimitExceededException& e) {
-            InternalError(NYql::NDqProto::StatusIds::OVERLOADED, TIssuesIds::KIKIMR_PRECONDITION_FAILED, TStringBuilder()
-                << "Mkql memory limit exceeded, limit: " << GetMkqlMemoryLimit()
-                << ", host: " << HostName()
-                << ", canAllocateExtraMemory: " << CanAllocateExtraMemory);
+            OnMemoryLimitExceptionHandler();
         } catch (const std::exception& e) {
             InternalError(NYql::NDqProto::StatusIds::INTERNAL_ERROR, TIssuesIds::UNEXPECTED, e.what());
         }
@@ -225,10 +222,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             TComputeActorClass* self = static_cast<TComputeActorClass*>(this);
             (self->*FuncBody)(ev);
         } catch (const NKikimr::TMemoryLimitExceededException& e) {
-            InternalError(NYql::NDqProto::StatusIds::OVERLOADED, TIssuesIds::KIKIMR_PRECONDITION_FAILED, TStringBuilder()
-                << "Mkql memory limit exceeded, limit: " << GetMkqlMemoryLimit()
-                << ", host: " << HostName()
-                << ", canAllocateExtraMemory: " << CanAllocateExtraMemory);
+            OnMemoryLimitExceptionHandler();
         } catch (const std::exception& e) {
             if (PassExceptions) {
                 throw;
@@ -316,6 +310,20 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
 
     virtual bool DoHandleChannelsAfterFinishImpl() = 0;
 
+    void OnMemoryLimitExceptionHandler() {
+        TString memoryConsumptionDetails = MemoryLimits.MemoryQuotaManager->MemoryConsumptionDetails();
+        TStringBuilder failureReason = TStringBuilder()
+            << "Mkql memory limit exceeded, limit: " << GetMkqlMemoryLimit()
+            << ", host: " << HostName()
+            << ", canAllocateExtraMemory: " << CanAllocateExtraMemory;
+
+        if (!memoryConsumptionDetails.empty()) {
+            failureReason << ", memory manager details: " << memoryConsumptionDetails;
+        }
+
+        InternalError(NYql::NDqProto::StatusIds::OVERLOADED, TIssuesIds::KIKIMR_PRECONDITION_FAILED, failureReason);
+    }
+
     void ProcessOutputsImpl(ERunStatus status) {
         ProcessOutputsState.LastRunStatus = status;
 
@@ -652,7 +660,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         OutputTransformsMap.at(outputIndex).FinishIsAcknowledged = true;
         ContinueExecute(EResumeSource::CATransformFinished);
     }
-    
+
 protected: //TDqComputeActorCheckpoints::ICallbacks
     //bool ReadyToCheckpoint() is pure and must be overriden in a derived class
 
@@ -1491,7 +1499,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             if (inputDesc.HasSource()) {
                 const auto watermarksMode = inputDesc.GetSource().GetWatermarksMode();
                 auto result = SourcesMap.emplace(
-                    i, 
+                    i,
                     static_cast<TDerived*>(this)->CreateInputHelper(LogPrefix, i, watermarksMode)
                 );
                 YQL_ENSURE(result.second);
@@ -1669,6 +1677,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             } else {
                 // in basic mode enum sources directly
                 for (auto& [inputIndex, sourceInfo] : SourcesMap) {
+                    if (!sourceInfo.AsyncInput)
+                        continue;
+
                     const auto& ingressStats = sourceInfo.AsyncInput->GetIngressStats();
                     ingressBytes += ingressStats.Bytes;
                     // ingress rows are usually not reported, so we count rows in task runner input
@@ -1691,6 +1702,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
 
             for (auto& [outputIndex, sinkInfo] : SinksMap) {
                 if (auto* sink = GetSink(outputIndex, sinkInfo)) {
+                    if (!sinkInfo.AsyncOutput)
+                        continue;
+
                     const auto& egressStats = sinkInfo.AsyncOutput->GetEgressStats();
                     const auto& pushStats = sink->GetPushStats();
                     if (RuntimeSettings.CollectFull()) {

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

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