init (#1348)

3 files changed, 23 insertions, 0 deletions

diff --git a/ydb/library/yql/minikql/mkql_terminator.cpp b/ydb/library/yql/minikql/mkql_terminator.cpp
index d86ba4edaf..81375f8362 100644
--- a/ydb/library/yql/minikql/mkql_terminator.cpp
+++ b/ydb/library/yql/minikql/mkql_terminator.cpp
@@ -1,6 +1,8 @@
 #include "defs.h"
 #include "mkql_terminator.h"
 
+#include <util/string/builder.h>
+
 namespace NKikimr {
 
 namespace NMiniKQL {
@@ -18,6 +20,18 @@ TBindTerminator::~TBindTerminator()
     Terminator = PreviousTerminator;
 }
 
+TThrowingBindTerminator::TThrowingBindTerminator()
+    : TBindTerminator(this)
+{
+}
+
+void TThrowingBindTerminator::Terminate(const char* message) const {
+    TStringBuf reason = (message ? TStringBuf(message) : TStringBuf("(unknown)"));
+    TString fullMessage = TStringBuilder() <<
+        "Terminate was called, reason(" << reason.size() << "): " << reason << Endl;
+    ythrow yexception() << fullMessage;
+}
+
 [[noreturn]] void MKQLTerminate(const char* message) {
     if (const auto t = TBindTerminator::Terminator)
         t->Terminate(message);
diff --git a/ydb/library/yql/minikql/mkql_terminator.h b/ydb/library/yql/minikql/mkql_terminator.h
index 2453e88938..137e8bd11f 100644
--- a/ydb/library/yql/minikql/mkql_terminator.h
+++ b/ydb/library/yql/minikql/mkql_terminator.h
@@ -26,6 +26,11 @@ private:
     ITerminator* PreviousTerminator;
 };
 
+struct TThrowingBindTerminator : public TBindTerminator, public ITerminator {
+    TThrowingBindTerminator();
+    void Terminate(const char* message) const final;
+};
+
 [[noreturn]] void MKQLTerminate(const char* message);
 
 }
diff --git a/ydb/library/yql/providers/dq/actors/proto_builder.cpp b/ydb/library/yql/providers/dq/actors/proto_builder.cpp
index 9a236f6edd..072eb7af5f 100644
--- a/ydb/library/yql/providers/dq/actors/proto_builder.cpp
+++ b/ydb/library/yql/providers/dq/actors/proto_builder.cpp
@@ -60,6 +60,7 @@ bool TProtoBuilder::CanBuildResultSet() const {
 }
 
 TString TProtoBuilder::BuildYson(TVector<NYql::NDq::TDqSerializedBatch>&& rows, ui64 maxBytesLimit) {
+    TThrowingBindTerminator t;
     ui64 size = 0;
     TStringStream out;
     NYson::TYsonWriter writer((IOutputStream*)&out);
@@ -82,6 +83,7 @@ TString TProtoBuilder::BuildYson(TVector<NYql::NDq::TDqSerializedBatch>&& rows,
 }
 
 bool TProtoBuilder::WriteYsonData(NYql::NDq::TDqSerializedBatch&& data, const std::function<bool(const TString& rawYson)>& func) {
+    TThrowingBindTerminator t;
     return WriteData(std::move(data), [&](const NYql::NUdf::TUnboxedValuePod& value) {
         auto rowYson = NCommon::WriteYsonValue(value, ResultType, ColumnOrder.empty() ? nullptr : &ColumnOrder);
         return func(rowYson);
@@ -89,6 +91,7 @@ bool TProtoBuilder::WriteYsonData(NYql::NDq::TDqSerializedBatch&& data, const st
 }
 
 bool TProtoBuilder::WriteData(NYql::NDq::TDqSerializedBatch&& data, const std::function<bool(const NYql::NUdf::TUnboxedValuePod& value)>& func) {
+    TThrowingBindTerminator t;
     TGuard<TScopedAlloc> allocGuard(Alloc);
 
     TMemoryUsageInfo memInfo("ProtoBuilder");
@@ -106,6 +109,7 @@ bool TProtoBuilder::WriteData(NYql::NDq::TDqSerializedBatch&& data, const std::f
 }
 
 bool TProtoBuilder::WriteData(TVector<NYql::NDq::TDqSerializedBatch>&& rows, const std::function<bool(const NYql::NUdf::TUnboxedValuePod& value)>& func) {
+    TThrowingBindTerminator t;
     TGuard<TScopedAlloc> allocGuard(Alloc);
 
     TMemoryUsageInfo memInfo("ProtoBuilder");