Merge 203246c into 82b202c

Author: GrigoriyPA

ydb/core/fq/libs/compute/ydb/ydb_connector_actor.cpp

@@ -130,6 +130,7 @@ class TYdbConnectorActor : public NActors::TActorBootstrapped<TYdbConnectorActor
     void Handle(const TEvYdbCompute::TEvFetchScriptResultRequest::TPtr& ev) {
         NYdb::NQuery::TFetchScriptResultsSettings settings;
         settings.FetchToken(ev->Get()->FetchToken);
+        settings.RowsLimit(0);
         QueryClient
             ->FetchScriptResults(ev->Get()->OperationId, ev->Get()->ResultSetId, settings)
             .Apply([actorSystem = NActors::TActivationContext::ActorSystem(), recipient = ev->Sender, cookie = ev->Cookie, database = ComputeConnection.database()](auto future) {

ydb/core/grpc_services/query/rpc_fetch_script_results.cpp

@@ -25,6 +25,7 @@ using TEvFetchScriptResultsRequest = TGrpcRequestNoOperationCall<Ydb::Query::Fet
     Ydb::Query::FetchScriptResultsResponse>;
 
 constexpr i64 MAX_ROWS_LIMIT = 1000;
+constexpr i64 SIZE_LIMIT = 20_MB;
 
 class TFetchScriptResultsRPC : public TRpcRequestActor<TFetchScriptResultsRPC, TEvFetchScriptResultsRequest, false> {
 public:
@@ -45,7 +46,7 @@ class TFetchScriptResultsRPC : public TRpcRequestActor<TFetchScriptResultsRPC, T
             return;
         }
 
-        if (req->rows_limit() <= 0) {
+        if (req->rows_limit() < 0) {
             Reply(Ydb::StatusIds::BAD_REQUEST, "Invalid rows limit");
             return;
         }
@@ -70,7 +71,7 @@ class TFetchScriptResultsRPC : public TRpcRequestActor<TFetchScriptResultsRPC, T
             return;
         }
 
-        Register(NKqp::CreateGetScriptExecutionResultActor(SelfId(), DatabaseName, ExecutionId, req->result_set_index(), RowsOffset, req->rows_limit() + 1));
+        Register(NKqp::CreateGetScriptExecutionResultActor(SelfId(), DatabaseName, ExecutionId, req->result_set_index(), RowsOffset, req->rows_limit(), SIZE_LIMIT, Request->GetDeadline()));
 
         Become(&TFetchScriptResultsRPC::StateFunc);
     }
@@ -80,18 +81,21 @@ class TFetchScriptResultsRPC : public TRpcRequestActor<TFetchScriptResultsRPC, T
         hFunc(NKqp::TEvKqp::TEvFetchScriptResultsResponse, Handle);
     )
 
-    void Handle(NKqp::TEvKqp::TEvFetchScriptResultsResponse::TPtr& ev) {
+    void Handle(NKqp::TEvFetchScriptResultsResponse::TPtr& ev) {
         Ydb::Query::FetchScriptResultsResponse resp;
-        resp.set_status(ev->Get()->Record.GetStatus());
-        resp.mutable_issues()->Swap(ev->Get()->Record.MutableIssues());
-        resp.set_result_set_index(static_cast<i64>(ev->Get()->Record.GetResultSetIndex()));
-        if (ev->Get()->Record.HasResultSet()) {
-            resp.mutable_result_set()->Swap(ev->Get()->Record.MutableResultSet());
-
-            const auto* userReq = GetProtoRequest();
-            if (resp.mutable_result_set()->rows_size() == userReq->rows_limit() + 1) {
-                resp.mutable_result_set()->mutable_rows()->DeleteSubrange(userReq->rows_limit(), 1);
-                resp.set_next_fetch_token(ToString(RowsOffset + userReq->rows_limit()));
+        resp.set_status(ev->Get()->Status);
+        resp.set_result_set_index(static_cast<i64>(GetProtoRequest()->result_set_index()));
+        if (ev->Get()->Issues) {
+            NYql::TIssue root;
+            for (const NYql::TIssue& issue : ev->Get()->Issues) {
+                root.AddSubIssue(MakeIntrusive<NYql::TIssue>(issue));
+            }
+            NYql::IssueToMessage(root, resp.mutable_issues());
+        }
+        if (ev->Get()->ResultSet) {
+            resp.mutable_result_set()->Swap(&(*ev->Get()->ResultSet));
+            if (ev->Get()->HasMoreResults) {
+                resp.set_next_fetch_token(ToString(RowsOffset + resp.result_set().rows_size()));
             }
         }
         Reply(resp.status(), std::move(resp));

ydb/core/kqp/common/events/events.h

@@ -142,9 +142,6 @@ struct TEvKqp {
 
     using TEvAbortExecution = NYql::NDq::TEvDq::TEvAbortExecution;
 
-    struct TEvFetchScriptResultsResponse : public TEventPB<TEvFetchScriptResultsResponse, NKikimrKqp::TEvFetchScriptResultsResponse, TKqpEvents::EvFetchScriptResultsResponse> {
-    };
-
     struct TEvCancelScriptExecutionRequest : public TEventPB<TEvCancelScriptExecutionRequest, NKikimrKqp::TEvCancelScriptExecutionRequest, TKqpEvents::EvCancelScriptExecutionRequest> {
     };
 

ydb/core/kqp/common/events/script_executions.h

@@ -224,15 +224,18 @@ struct TEvSaveScriptResultFinished : public NActors::TEventLocal<TEvSaveScriptRe
     NYql::TIssues Issues;
 };
 
-struct TEvFetchScriptResultsQueryResponse : public NActors::TEventLocal<TEvFetchScriptResultsQueryResponse, TKqpScriptExecutionEvents::EvFetchScriptResultsQueryResponse> {
-    TEvFetchScriptResultsQueryResponse(bool truncated, NKikimrKqp::TEvFetchScriptResultsResponse&& results)
-        : Truncated(truncated)
-        , Results(std::move(results))
-    {
-    }
+struct TEvFetchScriptResultsResponse : public NActors::TEventLocal<TEvFetchScriptResultsResponse, TKqpScriptExecutionEvents::EvFetchScriptResultsResponse> {
+    TEvFetchScriptResultsResponse(Ydb::StatusIds::StatusCode status, std::optional<Ydb::ResultSet>&& resultSet, bool hasMoreResults, NYql::TIssues issues)
+        : Status(status)
+        , ResultSet(std::move(resultSet))
+        , HasMoreResults(hasMoreResults)
+        , Issues(std::move(issues))
+    {}
 
-    bool Truncated;
-    NKikimrKqp::TEvFetchScriptResultsResponse Results;
+    Ydb::StatusIds::StatusCode Status;
+    std::optional<Ydb::ResultSet> ResultSet;
+    bool HasMoreResults;
+    NYql::TIssues Issues;
 };
 
 struct TEvSaveScriptExternalEffectRequest : public NActors::TEventLocal<TEvSaveScriptExternalEffectRequest, TKqpScriptExecutionEvents::EvSaveScriptExternalEffectRequest> {

ydb/core/kqp/common/simple/kqp_event_ids.h

@@ -148,7 +148,7 @@ struct TKqpScriptExecutionEvents {
         EvSaveScriptResultFinished,
         EvCheckAliveRequest,
         EvCheckAliveResponse,
-        EvFetchScriptResultsQueryResponse,
+        EvFetchScriptResultsResponse,
         EvSaveScriptExternalEffectRequest,
         EvSaveScriptExternalEffectResponse,
         EvScriptFinalizeRequest,