Better spilling errors (#8812)

Author: lll-phill-lll

ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp

@@ -135,7 +135,12 @@ Y_UNIT_TEST(HandleErrorsCorrectly) {
     Cerr << planres.GetStats()->GetAst() << Endl;
 
     auto result = db.ExecuteQuery(SimpleGraceJoinWithSpillingQuery, NYdb::NQuery::TTxControl::BeginTx().CommitTx(), NYdb::NQuery::TExecuteQuerySettings()).ExtractValueSync();
-    UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::INTERNAL_ERROR, result.GetIssues().ToString());
+    const auto errorMsg = result.GetIssues().ToString();
+    UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::INTERNAL_ERROR, errorMsg);
+
+    const auto spillingPrefix = "[Compute spilling]";
+    const auto pos = errorMsg.find(spillingPrefix);
+    UNIT_ASSERT_VALUES_UNEQUAL_C(pos, std::string::npos, "Spilling prefix not found in error message");
 }
 
 Y_UNIT_TEST(SelfJoinQueryService) {

ydb/library/yql/dq/actors/spilling/channel_storage_actor.cpp

@@ -82,7 +82,7 @@ class TDqChannelStorageActor : public IDqChannelStorageActor,
         if (!ErrorCallback_) Y_ABORT("Error: %s", error.c_str());
 
         LOG_E("Error: " << error);
-        ErrorCallback_(error);
+        ErrorCallback_(TStringBuilder() << "[Channel spilling]" << error);
         SendInternal(SpillingActorId_, new TEvents::TEvPoison);
         PassAway();
     }

ydb/library/yql/dq/actors/spilling/compute_storage_actor.cpp

@@ -76,7 +76,7 @@ class TDqComputeStorageActor : public NActors::TActorBootstrapped<TDqComputeStor
         if (!ErrorCallback_) Y_ABORT("Error: %s", error.c_str());
 
         LOG_E("Error: " << error);
-        ErrorCallback_(error);
+        ErrorCallback_(TStringBuilder() << "[Compute spilling]" << error);
         SendInternal(SpillingActorId_, new TEvents::TEvPoison);
         PassAway();
     }

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

@@ -1,6 +1,7 @@
 #include "spilling.h"
 #include "spilling_file.h"
 
+#include <format>
 #include <ydb/library/services/services.pb.h>
 #include <ydb/library/yql/utils/yql_panic.h>
 
@@ -406,7 +407,10 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             LOG_E("[Write] File size limit exceeded. "
                 << "From: " << ev->Sender << ", blobId: " << msg.BlobId << ", bytes: " << msg.Blob.size());
 
-            Send(ev->Sender, new TEvDqSpilling::TEvError("File size limit exceeded"));
+            const auto usedMb = (fd.TotalSize + msg.Blob.size()) / 1024 / 1024;
+            const auto limitMb = Config_.MaxFileSize / 1024 / 1024;
+
+            Send(ev->Sender, new TEvDqSpilling::TEvError(std::format("File size limit exceeded: {}/{}Mb", usedMb, limitMb)));
 
             Counters_->SpillingTooBigFileErrors->Inc();
             return;
@@ -416,7 +420,9 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             LOG_E("[Write] Total size limit exceeded. "
                 << "From: " << ev->Sender << ", blobId: " << msg.BlobId << ", bytes: " << msg.Blob.size());
 
-            Send(ev->Sender, new TEvDqSpilling::TEvError("Total size limit exceeded"));
+            const auto usedMb = (TotalSize_ + msg.Blob.size()) / 1024 / 1024;
+            const auto limitMb = Config_.MaxTotalSize / 1024 / 1024;
+            Send(ev->Sender, new TEvDqSpilling::TEvError(std::format("Total size limit exceeded: {}/{}Mb", usedMb, limitMb)));
 
             Counters_->SpillingNoSpaceErrors->Inc();
             return;

ydb/library/yql/dq/actors/spilling/spilling_file_ut.cpp

@@ -270,7 +270,7 @@ Y_UNIT_TEST_SUITE(DqSpillingFileTests) {
             runtime.Send(new IEventHandle(spillingActor, tester, ev));
 
             auto resp = runtime.GrabEdgeEvent<TEvDqSpilling::TEvError>(tester);
-            UNIT_ASSERT_STRINGS_EQUAL("Total size limit exceeded", resp->Get()->Message);
+            UNIT_ASSERT_STRINGS_EQUAL("Total size limit exceeded: 0/0Mb", resp->Get()->Message);
         }
     }
 
@@ -297,7 +297,7 @@ Y_UNIT_TEST_SUITE(DqSpillingFileTests) {
             runtime.Send(new IEventHandle(spillingActor, tester, ev));
 
             auto resp = runtime.GrabEdgeEvent<TEvDqSpilling::TEvError>(tester);
-            UNIT_ASSERT_STRINGS_EQUAL("File size limit exceeded", resp->Get()->Message);
+            UNIT_ASSERT_STRINGS_EQUAL("File size limit exceeded: 0/0Mb", resp->Get()->Message);
         }
     }