switch close session to public api

3 files changed, 24 insertions, 36 deletions

diff --git a/ydb/core/tx/datashard/datashard_ut_common_kqp.h b/ydb/core/tx/datashard/datashard_ut_common_kqp.h
index fb8eb59a6cc..32b7c8e8ad1 100644
--- a/ydb/core/tx/datashard/datashard_ut_common_kqp.h
+++ b/ydb/core/tx/datashard/datashard_ut_common_kqp.h
@@ -14,6 +14,9 @@ namespace NKqpHelpers {
     using TEvCreateSessionRequest = NKikimr::NGRpcService::TGrpcRequestOperationCall<Ydb::Table::CreateSessionRequest,
         Ydb::Table::CreateSessionResponse>;
 
+    using TEvDeleteSessionRequest = NKikimr::NGRpcService::TGrpcRequestOperationCall<Ydb::Table::DeleteSessionRequest,
+        Ydb::Table::DeleteSessionResponse>;
+
     template<class TResp>
     inline TResp AwaitResponse(TTestActorRuntime& runtime, NThreading::TFuture<TResp> f) {
         size_t responses = 0;
@@ -181,12 +184,11 @@ namespace NKqpHelpers {
         return runtime.GrabEdgeEventRethrow<NKqp::TEvKqp::TEvQueryResponse>(sender);
     }
 
-    inline void CloseSession(TTestActorRuntime& runtime, TActorId sender, const TString& sessionId) {
-        auto request = MakeHolder<NKqp::TEvKqp::TEvCloseSessionRequest>();
-        request->Record.MutableRequest()->SetSessionId(sessionId);
-        runtime.Send(
-            new IEventHandle(NKqp::MakeKqpProxyID(runtime.GetNodeId()), sender, request.Release()),
-            0, /* via actor system */ true);
+    inline void CloseSession(TTestActorRuntime& runtime, const TString& sessionId) {
+        Ydb::Table::DeleteSessionRequest request;
+        request.set_session_id(sessionId);
+        auto future = NRpcService::DoLocalRpc<TEvDeleteSessionRequest>(
+            std::move(request), "", "", /* token */ runtime.GetActorSystem(0));
     }
 
     inline THolder<NKqp::TEvKqp::TEvQueryRequest> MakeStreamRequest(
@@ -204,6 +206,14 @@ namespace NKqpHelpers {
         return request;
     }
 
+    inline TString FormatResult(const Ydb::Table::ExecuteQueryResult& result) {
+        if (result.result_sets_size() == 0) {
+            return "<empty>";
+        }
+        Cerr << JoinSeq(", ", result.result_sets(0).rows());
+        return JoinSeq(", ", result.result_sets(0).rows());
+    }
+
     inline TString KqpSimpleExec(TTestActorRuntime& runtime, const TString& query, bool staleRo = false) {
         TString sessionId = CreateSessionRPC(runtime);
         TString txId;
@@ -213,11 +223,7 @@ namespace NKqpHelpers {
         }
         Ydb::Table::ExecuteQueryResult result;
         response.operation().result().UnpackTo(&result);
-        if (result.result_sets_size() == 0) {
-            return "<empty>";
-        }
-        UNIT_ASSERT_VALUES_EQUAL(result.result_sets_size(), 1u);
-        return JoinSeq(", ", result.result_sets(0).rows());
+        return FormatResult(result);
     }
 
     inline TString KqpSimpleStaleRoExec(TTestActorRuntime& runtime, const TString& query) {
@@ -234,11 +240,7 @@ namespace NKqpHelpers {
         Ydb::Table::ExecuteQueryResult result;
         response.operation().result().UnpackTo(&result);
         txId = result.tx_meta().id();
-        if (result.result_sets_size() == 0) {
-            return "<empty>";
-        }
-        UNIT_ASSERT_VALUES_EQUAL(result.result_sets_size(), 1u);
-        return JoinSeq(", ", result.result_sets(0).rows());
+        return FormatResult(result);
     }
 
     inline TString KqpSimpleContinue(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query) {
@@ -250,12 +252,7 @@ namespace NKqpHelpers {
         Ydb::Table::ExecuteQueryResult result;
         response.operation().result().UnpackTo(&result);
         Y_VERIFY(result.tx_meta().id() == txId);
-        if (result.result_sets_size() == 0) {
-            return "<empty>";
-        }
-        UNIT_ASSERT_VALUES_EQUAL(result.result_sets_size(), 1u);
-        Cerr << JoinSeq(", ", result.result_sets(0).rows()) << Endl;
-        return JoinSeq(", ", result.result_sets(0).rows());
+        return FormatResult(result);
     }
 
     inline TString KqpSimpleCommit(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query) {
@@ -267,12 +264,7 @@ namespace NKqpHelpers {
         Ydb::Table::ExecuteQueryResult result;
         response.operation().result().UnpackTo(&result);
         Y_VERIFY(result.tx_meta().id().empty(), "must be empty transaction");
-        if (result.result_sets_size() == 0) {
-            return "<empty>";
-        }
-        UNIT_ASSERT_VALUES_EQUAL(result.result_sets_size(), 1u);
-        Cerr << JoinSeq(", ", result.result_sets(0).rows()) << Endl;
-        return JoinSeq(", ", result.result_sets(0).rows());
+        return FormatResult(result);
     }
 
 } // namespace NKqpHelpers
diff --git a/ydb/core/tx/datashard/datashard_ut_order.cpp b/ydb/core/tx/datashard/datashard_ut_order.cpp
index 02725bea9b7..7ae83116ade 100644
--- a/ydb/core/tx/datashard/datashard_ut_order.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_order.cpp
@@ -4662,10 +4662,9 @@ Y_UNIT_TEST_TWIN(TestSnapshotReadPriority, UnprotectedReads) {
     };
 
     auto execSnapshotRequest = [&](const TString& query) -> TString {
-        auto reqSender = runtime.AllocateEdgeActor();
         TString sessionId, txId;
         TString result = beginSnapshotRequest(sessionId, txId, query);
-        CloseSession(runtime, reqSender, sessionId);
+        CloseSession(runtime, sessionId);
         return result;
     };
 
@@ -5057,10 +5056,9 @@ Y_UNIT_TEST(TestUnprotectedReadsThenWriteVisibility) {
     };
 
     auto execSnapshotRequest = [&](const TString& query) -> TString {
-        auto reqSender = runtime.AllocateEdgeActor();
         TString sessionId, txId;
         TString result = beginSnapshotRequest(sessionId, txId, query);
-        CloseSession(runtime, reqSender, sessionId);
+        CloseSession(runtime, sessionId);
         return result;
     };
 
diff --git a/ydb/core/tx/datashard/datashard_ut_snapshot.cpp b/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
index c46edcfae4a..642483c4f97 100644
--- a/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_snapshot.cpp
@@ -1030,10 +1030,9 @@ Y_UNIT_TEST_SUITE(DataShardSnapshots) {
         };
 
         auto execSnapshotRequest = [&](const TString& query) -> TString {
-            auto reqSender = runtime.AllocateEdgeActor();
             TString sessionId, txId;
             TString result = beginSnapshotRequest(sessionId, txId, query);
-            CloseSession(runtime, reqSender, sessionId);
+            CloseSession(runtime, sessionId);
             return result;
         };
 
@@ -1145,10 +1144,9 @@ Y_UNIT_TEST_SUITE(DataShardSnapshots) {
         };
 
         auto execSnapshotRequest = [&](const TString& query) -> TString {
-            auto reqSender = runtime.AllocateEdgeActor();
             TString sessionId, txId;
             TString result = beginSnapshotRequest(sessionId, txId, query);
-            CloseSession(runtime, reqSender, sessionId);
+            CloseSession(runtime, sessionId);
             return result;
         };