Invert grpc_request_proxy dependence for table service. KIKIMR-13646

ref:0b126dafc321557e232e9131321cc8ffa4790f9a

ydb/core/blobstorage/testload/test_load_kqp.cpp

@@ -223,7 +223,8 @@ private:
             }
         };
         using namespace NGRpcService;
-        using TEvCreateSessionRequest = TGRpcRequestWrapper<TRpcServices::EvCreateSession, Ydb::Table::CreateSessionRequest, Ydb::Table::CreateSessionResponse, true, TRateLimiterMode::Rps>;
+        using TEvCreateSessionRequest = TGrpcRequestOperationCall<Ydb::Table::CreateSessionRequest,
+            Ydb::Table::CreateSessionResponse>;
         NKikimr::NRpcService::DoLocalRpcSameMailbox<TEvCreateSessionRequest>(
             std::move(request), std::move(cb), WorkingDir, TString(), ctx
         );
@@ -247,7 +248,8 @@ private:
             }
         };
         using namespace NGRpcService;
-        using TEvExecuteSchemeQueryRequest = TGRpcRequestWrapper<TRpcServices::EvExecuteSchemeQuery, Ydb::Table::ExecuteSchemeQueryRequest, Ydb::Table::ExecuteSchemeQueryResponse, true, TRateLimiterMode::Rps>;
+        using TEvExecuteSchemeQueryRequest = TGrpcRequestOperationCall<Ydb::Table::ExecuteSchemeQueryRequest,
+            Ydb::Table::ExecuteSchemeQueryResponse>;
         NKikimr::NRpcService::DoLocalRpcSameMailbox<TEvExecuteSchemeQueryRequest>(
             std::move(request), std::move(cb), WorkingDir, TString(), ctx
         );

ydb/core/grpc_services/base/base.h

@@ -360,6 +360,11 @@ public:
     virtual void AddServerHint(const TString& hint) = 0;
     virtual void SetCostInfo(float consumed_units) = 0;
 
+    virtual void SetStreamingNotify(NGrpc::IRequestContextBase::TOnNextReply&& cb) = 0;
+    virtual void FinishStream() = 0;
+
+    virtual void SendSerializedResult(TString&& in, Ydb::StatusIds::StatusCode status) = 0;
+
 private:
     virtual void Reply(NProtoBuf::Message* resp, ui32 status = 0) = 0;
 };
@@ -378,6 +383,7 @@ public:
 };
 
 class IRequestNoOpCtx : public IRequestCtx {
+
 };
 
 struct TCommonResponseFillerImpl {
@@ -914,7 +920,7 @@ public:
         return GetPeerMetaValues(NYdb::YDB_REQUEST_TYPE_HEADER);
     }
 
-    void SendSerializedResult(TString&& in, Ydb::StatusIds::StatusCode status) {
+    void SendSerializedResult(TString&& in, Ydb::StatusIds::StatusCode status) override {
         // res->data() pointer is used inside grpc code.
         // So this object should be destroyed during grpc_slice destroying routine
         auto res = new TString;
@@ -954,9 +960,8 @@ public:
         return google::protobuf::Arena::CreateMessage<TResult>(ctx->GetArena());
     }
 
-    template<typename Tcb>
-    void SetStreamingNotify(Tcb&& cb) {
-        Ctx_->SetNextReplyCallback(cb);
+    void SetStreamingNotify(NGrpc::IRequestContextBase::TOnNextReply&& cb) override {
+        Ctx_->SetNextReplyCallback(std::move(cb));
     }
 
     void SetClientLostAction(std::function<void()>&& cb) override {
@@ -969,7 +974,7 @@ public:
         Ctx_->GetFinishFuture().Subscribe(std::move(shutdown));
     }
 
-    void FinishStream() {
+    void FinishStream() override {
         Ctx_->FinishStreamingOk();
     }
 
@@ -1033,6 +1038,8 @@ class TGrpcRequestCall
     typedef typename std::conditional<IsOperation, IRequestOpCtx, IRequestNoOpCtx>::type TRequestIface;
 public:
     static constexpr bool IsOp = IsOperation;
+    static IActor* CreateRpcActor(typename std::conditional<IsOperation, IRequestOpCtx, IRequestNoOpCtx>::type* msg);
+
     TGrpcRequestCall(NGrpc::IRequestContextBase* ctx,
         void (*cb)(std::unique_ptr<TRequestIface>, const IFacilityProvider&), TRequestAuxSettings auxSettings = {})
         : TGRpcRequestWrapperImpl<

ydb/core/grpc_services/grpc_request_proxy.cpp

@@ -566,36 +566,17 @@ void TGRpcRequestProxyImpl::StateFunc(TAutoPtr<IEventHandle>& ev, const TActorCo
     switch (ev->GetTypeRewrite()) {
         HFunc(TRefreshTokenImpl, PreHandle);
         HFunc(TEvLoginRequest, PreHandle);
-        HFunc(TEvAlterTableRequest, PreHandle);
-        HFunc(TEvCreateTableRequest, PreHandle);
-        HFunc(TEvDropTableRequest, PreHandle);
         HFunc(TEvGetOperationRequest, PreHandle);
         HFunc(TEvCancelOperationRequest, PreHandle);
         HFunc(TEvForgetOperationRequest, PreHandle);
         HFunc(TEvListOperationsRequest, PreHandle);
-        HFunc(TEvCreateSessionRequest, PreHandle);
-        HFunc(TEvKeepAliveRequest, PreHandle);
-        HFunc(TEvDeleteSessionRequest, PreHandle);
-        HFunc(TEvCopyTableRequest, PreHandle);
-        HFunc(TEvCopyTablesRequest, PreHandle);
-        HFunc(TEvRenameTablesRequest, PreHandle);
-        HFunc(TEvDescribeTableRequest, PreHandle);
-        HFunc(TEvReadTableRequest, PreHandle);
-        HFunc(TEvExplainDataQueryRequest, PreHandle);
-        HFunc(TEvPrepareDataQueryRequest, PreHandle);
-        HFunc(TEvExecuteDataQueryRequest, PreHandle);
-        HFunc(TEvExecuteSchemeQueryRequest, PreHandle);
         HFunc(TEvCreateTenantRequest, PreHandle);
         HFunc(TEvAlterTenantRequest, PreHandle);
         HFunc(TEvGetTenantStatusRequest, PreHandle);
         HFunc(TEvListTenantsRequest, PreHandle);
         HFunc(TEvRemoveTenantRequest, PreHandle);
-        HFunc(TEvBeginTransactionRequest, PreHandle);
-        HFunc(TEvCommitTransactionRequest, PreHandle);
-        HFunc(TEvRollbackTransactionRequest, PreHandle);
         HFunc(TEvListEndpointsRequest, PreHandle);
         HFunc(TEvDescribeTenantOptionsRequest, PreHandle);
-        HFunc(TEvDescribeTableOptionsRequest, PreHandle);
         HFunc(TEvCreateCoordinationNode, PreHandle);
         HFunc(TEvAlterCoordinationNode, PreHandle);
         HFunc(TEvDropCoordinationNode, PreHandle);
@@ -620,7 +601,6 @@ void TGRpcRequestProxyImpl::StateFunc(TAutoPtr<IEventHandle>& ev, const TActorCo
         HFunc(TEvImportFromS3Request, PreHandle);
         HFunc(TEvImportDataRequest, PreHandle);
         HFunc(TEvDiscoverPQClustersRequest, PreHandle);
-        HFunc(TEvBulkUpsertRequest, PreHandle);
         HFunc(TEvWhoAmIRequest, PreHandle);
         HFunc(TEvCreateRateLimiterResource, PreHandle);
         HFunc(TEvAlterRateLimiterResource, PreHandle);
@@ -632,7 +612,6 @@ void TGRpcRequestProxyImpl::StateFunc(TAutoPtr<IEventHandle>& ev, const TActorCo
         HFunc(TEvKikhouseRefreshSnapshotRequest, PreHandle);
         HFunc(TEvKikhouseDiscardSnapshotRequest, PreHandle);
         HFunc(TEvSelfCheckRequest, PreHandle);
-        HFunc(TEvStreamExecuteScanQueryRequest, PreHandle);
         HFunc(TEvCoordinationSessionRequest, PreHandle);
         HFunc(TEvLongTxBeginRequest, PreHandle);
         HFunc(TEvLongTxCommitRequest, PreHandle);

ydb/core/grpc_services/grpc_request_proxy.h

@@ -48,36 +48,17 @@ public:
     };
 
 protected:
-    void Handle(TEvAlterTableRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvCreateTableRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvDropTableRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvCopyTableRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvCopyTablesRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvRenameTablesRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvDescribeTableRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvGetOperationRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvCancelOperationRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvForgetOperationRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvListOperationsRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvCreateSessionRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvKeepAliveRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvDeleteSessionRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvReadTableRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvExplainDataQueryRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvPrepareDataQueryRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvExecuteDataQueryRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvExecuteSchemeQueryRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvCreateTenantRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvAlterTenantRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvGetTenantStatusRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvListTenantsRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvRemoveTenantRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvBeginTransactionRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvCommitTransactionRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvRollbackTransactionRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvListEndpointsRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvDescribeTenantOptionsRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvDescribeTableOptionsRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvCreateCoordinationNode::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvAlterCoordinationNode::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvDropCoordinationNode::TPtr& ev, const TActorContext& ctx);
@@ -102,7 +83,6 @@ protected:
     void Handle(TEvImportFromS3Request::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvImportDataRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvDiscoverPQClustersRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvBulkUpsertRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvWhoAmIRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvCreateRateLimiterResource::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvAlterRateLimiterResource::TPtr& ev, const TActorContext& ctx);
@@ -115,7 +95,6 @@ protected:
     void Handle(TEvKikhouseDiscardSnapshotRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvSelfCheckRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvLoginRequest::TPtr& ev, const TActorContext& ctx);
-    void Handle(TEvStreamExecuteScanQueryRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvCoordinationSessionRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvLongTxBeginRequest::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvLongTxCommitRequest::TPtr& ev, const TActorContext& ctx);

ydb/core/grpc_services/local_rpc/local_rpc.h

@@ -145,6 +145,21 @@ public:
         Y_FAIL("Unimplemented for local rpc");
     }
 
+    void SetStreamingNotify(NGrpc::IRequestContextBase::TOnNextReply&& cb) override {
+        Y_UNUSED(cb);
+        Y_FAIL("Unimplemented for local rpc");
+    }
+
+    void FinishStream() override {
+        Y_FAIL("Unimplemented for local rpc");
+    }
+
+    virtual void SendSerializedResult(TString&& in, Ydb::StatusIds::StatusCode status) override {
+        Y_UNUSED(in);
+        Y_UNUSED(status);
+        Y_FAIL("Unimplemented for local rpc");
+    }
+
     TMaybe<TString> GetTraceId() const override {
         return Nothing();
     }

ydb/core/grpc_services/rpc_alter_table.cpp

@@ -1,9 +1,11 @@
-#include "grpc_request_proxy.h"
+#include "service_table.h"
+#include <ydb/core/grpc_services/base/base.h>
 
 #include "rpc_scheme_base.h"
 #include "rpc_common.h"
 #include "operation_helpers.h"
 #include "table_settings.h"
+#include "service_table.h"
 
 #include <ydb/core/base/tablet_pipe.h>
 #include <ydb/core/cms/console/configs_dispatcher.h>
@@ -77,6 +79,9 @@ static std::pair<StatusIds::StatusCode, TString> CheckAddIndexDesc(const Ydb::Ta
     return {StatusIds::SUCCESS, ""};
 }
 
+using TEvAlterTableRequest = TGrpcRequestOperationCall<Ydb::Table::AlterTableRequest,
+    Ydb::Table::AlterTableResponse>;
+
 class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTableRequest> {
     using TBase = TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTableRequest>;
 
@@ -537,8 +542,8 @@ private:
     TTableProfiles Profiles;
 };
 
-void TGRpcRequestProxy::Handle(TEvAlterTableRequest::TPtr& ev, const TActorContext& ctx) {
-    ctx.Register(new TAlterTableRPC(ev->Release().Release()));
+void DoAlterTableRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider &) {
+    TActivationContext::AsActorContext().Register(new TAlterTableRPC(p.release()));
 }
 
 template<>
@@ -546,6 +551,5 @@ IActor* TEvAlterTableRequest::CreateRpcActor(NKikimr::NGRpcService::IRequestOpCt
     return new TAlterTableRPC(msg);
 }
 
-
 } // namespace NKikimr
 } // namespace NGRpcService

ydb/core/grpc_services/rpc_begin_transaction.cpp

@@ -1,8 +1,10 @@
-#include "grpc_request_proxy.h"
+#include "service_table.h"
+#include <ydb/core/grpc_services/base/base.h>
 
 #include "rpc_calls.h"
 #include "rpc_kqp_base.h"
 #include "rpc_common.h"
+#include "service_table.h"
 
 #include <ydb/library/yql/public/issue/yql_issue_message.h>
 #include <ydb/library/yql/public/issue/yql_issue.h>
@@ -14,13 +16,16 @@ using namespace NActors;
 using namespace Ydb;
 using namespace NKqp;
 
+using TEvBeginTransactionRequest = TGrpcRequestOperationCall<Ydb::Table::BeginTransactionRequest,
+    Ydb::Table::BeginTransactionResponse>;
+
 class TBeginTransactionRPC : public TRpcKqpRequestActor<TBeginTransactionRPC, TEvBeginTransactionRequest> {
     using TBase = TRpcKqpRequestActor<TBeginTransactionRPC, TEvBeginTransactionRequest>;
 
 public:
     using TResult = Ydb::Table::BeginTransactionResult;
 
-    TBeginTransactionRPC(TEvBeginTransactionRequest* msg)
+    TBeginTransactionRPC(IRequestOpCtx* msg)
         : TBase(msg) {}
 
     void Bootstrap(const TActorContext& ctx) {
@@ -104,8 +109,8 @@ private:
     }
 };
 
-void TGRpcRequestProxy::Handle(TEvBeginTransactionRequest::TPtr& ev, const TActorContext& ctx) {
-    ctx.Register(new TBeginTransactionRPC(ev->Release().Release()));
+void DoBeginTransactionRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider &) {
+    TActivationContext::AsActorContext().Register(new TBeginTransactionRPC(p.release()));
 }
 
 } // namespace NGRpcService

ydb/core/grpc_services/rpc_calls.h

@@ -45,28 +45,10 @@ void FillYdbStatus(Draft::Dummy::PingResponse& resp, const NYql::TIssues& issues
 template <>
 void FillYdbStatus(Ydb::Coordination::SessionResponse& resp, const NYql::TIssues& issues, Ydb::StatusIds::StatusCode status);
 
-using TEvAlterTableRequest = TGRpcRequestValidationWrapper<TRpcServices::EvAlterTable, Ydb::Table::AlterTableRequest, Ydb::Table::AlterTableResponse, true, TRateLimiterMode::Rps>;
-using TEvCreateTableRequest = TGRpcRequestValidationWrapper<TRpcServices::EvCreateTable, Ydb::Table::CreateTableRequest, Ydb::Table::CreateTableResponse, true, TRateLimiterMode::Rps>;
-using TEvDropTableRequest = TGRpcRequestWrapper<TRpcServices::EvDropTable, Ydb::Table::DropTableRequest, Ydb::Table::DropTableResponse, true, TRateLimiterMode::Rps>;
-using TEvCopyTableRequest = TGRpcRequestWrapper<TRpcServices::EvCopyTable, Ydb::Table::CopyTableRequest, Ydb::Table::CopyTableResponse, true, TRateLimiterMode::Rps>;
-using TEvCopyTablesRequest = TGRpcRequestWrapper<TRpcServices::EvCopyTables, Ydb::Table::CopyTablesRequest, Ydb::Table::CopyTablesResponse, true, TRateLimiterMode::Rps>;
-using TEvRenameTablesRequest = TGRpcRequestWrapper<TRpcServices::EvRenameTables, Ydb::Table::RenameTablesRequest, Ydb::Table::RenameTablesResponse, true, TRateLimiterMode::Rps>;
-using TEvDescribeTableRequest = TGRpcRequestWrapper<TRpcServices::EvDescribeTable, Ydb::Table::DescribeTableRequest, Ydb::Table::DescribeTableResponse, true, TRateLimiterMode::Rps>;
 using TEvGetOperationRequest = TGRpcRequestValidationWrapper<TRpcServices::EvGetOperation, Ydb::Operations::GetOperationRequest, Ydb::Operations::GetOperationResponse, true, TRateLimiterMode::Rps>;
 using TEvCancelOperationRequest = TGRpcRequestValidationWrapper<TRpcServices::EvCancelOperation, Ydb::Operations::CancelOperationRequest, Ydb::Operations::CancelOperationResponse, false, TRateLimiterMode::Rps>;
 using TEvForgetOperationRequest = TGRpcRequestValidationWrapper<TRpcServices::EvForgetOperation, Ydb::Operations::ForgetOperationRequest, Ydb::Operations::ForgetOperationResponse, false, TRateLimiterMode::Rps>;
 using TEvListOperationsRequest = TGRpcRequestValidationWrapper<TRpcServices::EvListOperations, Ydb::Operations::ListOperationsRequest, Ydb::Operations::ListOperationsResponse, false, TRateLimiterMode::Rps>;
-using TEvCreateSessionRequest = TGRpcRequestWrapper<TRpcServices::EvCreateSession, Ydb::Table::CreateSessionRequest, Ydb::Table::CreateSessionResponse, true, TRateLimiterMode::Rps>;
-using TEvDeleteSessionRequest = TGRpcRequestWrapper<TRpcServices::EvDeleteSession, Ydb::Table::DeleteSessionRequest, Ydb::Table::DeleteSessionResponse, true>;
-using TEvKeepAliveRequest = TGRpcRequestWrapper<TRpcServices::EvKeepAlive, Ydb::Table::KeepAliveRequest, Ydb::Table::KeepAliveResponse, true, TRateLimiterMode::Rps>;
-using TEvReadTableRequest = TGRpcRequestWrapper<TRpcServices::EvReadTable, Ydb::Table::ReadTableRequest, Ydb::Table::ReadTableResponse, false, TRateLimiterMode::RuOnProgress>;
-using TEvExplainDataQueryRequest = TGRpcRequestWrapper<TRpcServices::EvExplainDataQuery, Ydb::Table::ExplainDataQueryRequest, Ydb::Table::ExplainDataQueryResponse, true, TRateLimiterMode::Rps>;
-using TEvPrepareDataQueryRequest = TGRpcRequestWrapper<TRpcServices::EvPrepareDataQuery, Ydb::Table::PrepareDataQueryRequest, Ydb::Table::PrepareDataQueryResponse, true, TRateLimiterMode::Ru>;
-using TEvExecuteDataQueryRequest = TGRpcRequestWrapper<TRpcServices::EvExecuteDataQuery, Ydb::Table::ExecuteDataQueryRequest, Ydb::Table::ExecuteDataQueryResponse, true, TRateLimiterMode::Ru>;
-using TEvExecuteSchemeQueryRequest = TGRpcRequestWrapper<TRpcServices::EvExecuteSchemeQuery, Ydb::Table::ExecuteSchemeQueryRequest, Ydb::Table::ExecuteSchemeQueryResponse, true, TRateLimiterMode::Rps>;
-using TEvBeginTransactionRequest = TGRpcRequestWrapper<TRpcServices::EvBeginTransaction, Ydb::Table::BeginTransactionRequest, Ydb::Table::BeginTransactionResponse, true, TRateLimiterMode::Rps>;
-using TEvCommitTransactionRequest = TGRpcRequestWrapper<TRpcServices::EvCommitTransaction, Ydb::Table::CommitTransactionRequest, Ydb::Table::CommitTransactionResponse, true>;
-using TEvRollbackTransactionRequest = TGRpcRequestWrapper<TRpcServices::EvRollbackTransaction, Ydb::Table::RollbackTransactionRequest, Ydb::Table::RollbackTransactionResponse, true>;
 using TEvCreateTenantRequest = TGRpcRequestWrapper<TRpcServices::EvCreateTenant, Ydb::Cms::CreateDatabaseRequest, Ydb::Cms::CreateDatabaseResponse, true>;
 using TEvAlterTenantRequest = TGRpcRequestWrapper<TRpcServices::EvAlterTenant, Ydb::Cms::AlterDatabaseRequest, Ydb::Cms::AlterDatabaseResponse, true>;
 using TEvGetTenantStatusRequest = TGRpcRequestWrapper<TRpcServices::EvGetTenantStatus, Ydb::Cms::GetDatabaseStatusRequest, Ydb::Cms::GetDatabaseStatusResponse, true>;
@@ -74,7 +56,6 @@ using TEvListTenantsRequest = TGRpcRequestWrapper<TRpcServices::EvListTenants, Y
 using TEvRemoveTenantRequest = TGRpcRequestWrapper<TRpcServices::EvRemoveTenant, Ydb::Cms::RemoveDatabaseRequest, Ydb::Cms::RemoveDatabaseResponse, true>;
 using TEvListEndpointsRequest = TGRpcRequestWrapper<TRpcServices::EvListEndpoints, Ydb::Discovery::ListEndpointsRequest, Ydb::Discovery::ListEndpointsResponse, true>;
 using TEvDescribeTenantOptionsRequest = TGRpcRequestWrapper<TRpcServices::EvDescribeTenantOptions, Ydb::Cms::DescribeDatabaseOptionsRequest, Ydb::Cms::DescribeDatabaseOptionsResponse, true>;
-using TEvDescribeTableOptionsRequest = TGRpcRequestWrapper<TRpcServices::EvDescribeTableOptions, Ydb::Table::DescribeTableOptionsRequest, Ydb::Table::DescribeTableOptionsResponse, true, TRateLimiterMode::Rps>;
 using TEvCreateCoordinationNode = TGRpcRequestWrapper<TRpcServices::EvCreateCoordinationNode, Ydb::Coordination::CreateNodeRequest, Ydb::Coordination::CreateNodeResponse, true, TRateLimiterMode::Rps>;
 using TEvAlterCoordinationNode = TGRpcRequestWrapper<TRpcServices::EvAlterCoordinationNode, Ydb::Coordination::AlterNodeRequest, Ydb::Coordination::AlterNodeResponse, true, TRateLimiterMode::Rps>;
 using TEvDropCoordinationNode = TGRpcRequestWrapper<TRpcServices::EvDropCoordinationNode, Ydb::Coordination::DropNodeRequest, Ydb::Coordination::DropNodeResponse, true, TRateLimiterMode::Rps>;
@@ -99,7 +80,6 @@ using TEvExportToS3Request = TGRpcRequestValidationWrapper<TRpcServices::EvExpor
 using TEvImportFromS3Request = TGRpcRequestValidationWrapper<TRpcServices::EvImportFromS3, Ydb::Import::ImportFromS3Request, Ydb::Import::ImportFromS3Response, true>;
 using TEvImportDataRequest = TGRpcRequestValidationWrapper<TRpcServices::EvImportData, Ydb::Import::ImportDataRequest, Ydb::Import::ImportDataResponse, true>;
 using TEvDiscoverPQClustersRequest = TGRpcRequestWrapper<TRpcServices::EvDiscoverPQClusters, Ydb::PersQueue::ClusterDiscovery::DiscoverClustersRequest, Ydb::PersQueue::ClusterDiscovery::DiscoverClustersResponse, true>;
-using TEvBulkUpsertRequest = TGRpcRequestWrapper<TRpcServices::EvBulkUpsert, Ydb::Table::BulkUpsertRequest, Ydb::Table::BulkUpsertResponse, true, TRateLimiterMode::Ru>;
 using TEvWhoAmIRequest = TGRpcRequestWrapper<TRpcServices::EvWhoAmI, Ydb::Discovery::WhoAmIRequest, Ydb::Discovery::WhoAmIResponse, true, TRateLimiterMode::Rps>;
 using TEvCreateRateLimiterResource = TGRpcRequestWrapper<TRpcServices::EvCreateRateLimiterResource, Ydb::RateLimiter::CreateResourceRequest, Ydb::RateLimiter::CreateResourceResponse, true, TRateLimiterMode::Rps>;
 using TEvAlterRateLimiterResource = TGRpcRequestWrapper<TRpcServices::EvAlterRateLimiterResource, Ydb::RateLimiter::AlterResourceRequest, Ydb::RateLimiter::AlterResourceResponse, true, TRateLimiterMode::Rps>;
@@ -112,7 +92,6 @@ using TEvKikhouseRefreshSnapshotRequest = TGRpcRequestWrapper<TRpcServices::EvKi
 using TEvKikhouseDiscardSnapshotRequest = TGRpcRequestWrapper<TRpcServices::EvKikhouseDiscardSnapshot, Ydb::ClickhouseInternal::DiscardSnapshotRequest, Ydb::ClickhouseInternal::DiscardSnapshotResponse, true>;
 using TEvSelfCheckRequest = TGRpcRequestWrapper<TRpcServices::EvSelfCheck, Ydb::Monitoring::SelfCheckRequest, Ydb::Monitoring::SelfCheckResponse, true>;
 using TEvLoginRequest = TGRpcRequestWrapperNoAuth<TRpcServices::EvLogin, Ydb::Auth::LoginRequest, Ydb::Auth::LoginResponse>;
-using TEvStreamExecuteScanQueryRequest = TGRpcRequestWrapper<TRpcServices::EvStreamExecuteScanQuery, Ydb::Table::ExecuteScanQueryRequest, Ydb::Table::ExecuteScanQueryPartialResponse, false, TRateLimiterMode::RuOnProgress>;
 using TEvCoordinationSessionRequest = TGRpcRequestBiStreamWrapper<TRpcServices::EvCoordinationSession, Ydb::Coordination::SessionRequest, Ydb::Coordination::SessionResponse>;
 using TEvLongTxBeginRequest = TGRpcRequestWrapper<TRpcServices::EvLongTxBegin, Ydb::LongTx::BeginTransactionRequest, Ydb::LongTx::BeginTransactionResponse, true>;
 using TEvLongTxCommitRequest = TGRpcRequestWrapper<TRpcServices::EvLongTxCommit, Ydb::LongTx::CommitTransactionRequest, Ydb::LongTx::CommitTransactionResponse, true>;

ydb/core/grpc_services/rpc_commit_transaction.cpp

@@ -1,8 +1,10 @@
-#include "grpc_request_proxy.h"
+#include "service_table.h"
+#include <ydb/core/grpc_services/base/base.h>
 
 #include "rpc_calls.h"
 #include "rpc_kqp_base.h"
 #include "rpc_common.h"
+#include "service_table.h"
 
 #include <ydb/library/yql/public/issue/yql_issue_message.h>
 #include <ydb/library/yql/public/issue/yql_issue.h>
@@ -14,11 +16,14 @@ using namespace NActors;
 using namespace Ydb;
 using namespace NKqp;
 
+using TEvCommitTransactionRequest = TGrpcRequestOperationCall<Ydb::Table::CommitTransactionRequest,
+Ydb::Table::CommitTransactionResponse>;
+
 class TCommitTransactionRPC : public TRpcKqpRequestActor<TCommitTransactionRPC, TEvCommitTransactionRequest> {
     using TBase = TRpcKqpRequestActor<TCommitTransactionRPC, TEvCommitTransactionRequest>;
 
 public:
-    TCommitTransactionRPC(TEvCommitTransactionRequest* msg)
+    TCommitTransactionRPC(IRequestOpCtx* msg)
         : TBase(msg) {}
 
     void Bootstrap(const TActorContext& ctx) {
@@ -92,8 +97,8 @@ private:
     }
 };
 
-void TGRpcRequestProxy::Handle(TEvCommitTransactionRequest::TPtr& ev, const TActorContext& ctx) {
-    ctx.Register(new TCommitTransactionRPC(ev->Release().Release()));
+void DoCommitTransactionRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider &) {
+    TActivationContext::AsActorContext().Register(new TCommitTransactionRPC(p.release()));
 }
 
 } // namespace NGRpcService

ydb/core/grpc_services/rpc_common.h

@@ -8,6 +8,7 @@
 
 namespace NKikimr {
 namespace NGRpcService {
+class IRequestCtx;
 
 template<typename TEv>
 inline void SetRlPath(TEv& ev, const IRequestCtx& ctx) {