ExecuteData set transaction mode always (#7982)

Co-authored-by: Ivan Sukhov <[email protected]>

Author: StekPerepolnen

ydb/core/viewer/viewer_query.h

@@ -305,6 +305,10 @@ class TJsonQuery : public TViewerPipeClient {
             request.SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);
             request.SetKeepSession(false);
             SetTransactionMode(request);
+            if (!request.txcontrol().has_begin_tx()) {
+                request.mutable_txcontrol()->mutable_begin_tx()->mutable_serializable_read_write();
+                request.mutable_txcontrol()->set_commit_tx(true);
+            }
         } else if (Action == "explain" || Action == "explain-ast" || Action == "explain-data") {
             request.SetAction(NKikimrKqp::QUERY_ACTION_EXPLAIN);
             request.SetType(NKikimrKqp::QUERY_TYPE_SQL_DML);

ydb/core/viewer/viewer_ut.cpp

@@ -17,6 +17,7 @@
 
 #include <library/cpp/testing/unittest/registar.h>
 #include <library/cpp/testing/unittest/tests_data.h>
+#include <ydb/core/kqp/common/kqp.h>
 #include <ydb/core/testlib/test_client.h>
 #include <ydb/core/testlib/tenant_runtime.h>
 #include <ydb/public/lib/deprecated/kicli/kicli.h>
@@ -1593,6 +1594,91 @@ Y_UNIT_TEST_SUITE(Viewer) {
         size_t AuthorizeTicketFails = 0;
     };
 
+    TString PostQuery(TKeepAliveHttpClient& httpClient, TString query, TString action = "", TString transactionMode = "") {
+        TStringStream requestBody;
+        requestBody
+            << "{ \"query\": \"" << query << "\","
+            << " \"database\": \"/Root\","
+            << " \"action\": \"" << action << "\","
+            << " \"syntax\": \"yql_v1\","
+            << " \"transaction_mode\": \"" << transactionMode << "\","
+            << " \"stats\": \"none\" }";
+        TStringStream responseStream;
+        TKeepAliveHttpClient::THeaders headers;
+        headers["Content-Type"] = "application/json";
+        headers["Authorization"] = "test_ydb_token";
+        const TKeepAliveHttpClient::THttpCode statusCode = httpClient.DoPost("/viewer/query?timeout=600000&base64=false&schema=modern", requestBody.Str(), &responseStream, headers);
+        const TString response = responseStream.ReadAll();
+        UNIT_ASSERT_EQUAL_C(statusCode, HTTP_OK, statusCode << ": " << response);
+        return response;
+    }
+
+    Y_UNIT_TEST(ExecuteQueryDoesntExecuteSchemeOperationsInsideTransation) {
+        TPortManager tp;
+        ui16 port = tp.GetPort(2134);
+        ui16 grpcPort = tp.GetPort(2135);
+        ui16 monPort = tp.GetPort(8765);
+        auto settings = TServerSettings(port);
+        settings.InitKikimrRunConfig()
+                .SetNodeCount(1)
+                .SetUseRealThreads(true)
+                .SetDomainName("Root")
+                .SetMonitoringPortOffset(monPort, true);
+
+        TServer server(settings);
+        server.EnableGRpc(grpcPort);
+        TClient client(settings);
+        client.InitRootScheme();
+
+        TTestActorRuntime& runtime = *server.GetRuntime();
+        runtime.SetLogPriority(NKikimrServices::TICKET_PARSER, NLog::PRI_TRACE);
+
+        TKeepAliveHttpClient httpClient("localhost", monPort);
+
+        //Scheme operations cannot be executed inside transaction
+        TString response = PostQuery(httpClient, "CREATE TABLE `/Root/Test` (Key Uint64, Value String, PRIMARY KEY (Key));", "execute-query", "serializable-read-write");
+        {
+            NJson::TJsonReaderConfig jsonCfg;
+            NJson::TJsonValue json;
+            NJson::ReadJsonTree(response, &jsonCfg, &json, /* throwOnError = */ true);
+            UNIT_ASSERT_EQUAL_C(json["status"].GetString(), "PRECONDITION_FAILED", response);
+        }
+    }
+
+    Y_UNIT_TEST(UseTransactionWhenExecuteDataActionQuery) {
+        TPortManager tp;
+        ui16 port = tp.GetPort(2134);
+        ui16 grpcPort = tp.GetPort(2135);
+        ui16 monPort = tp.GetPort(8765);
+        auto settings = TServerSettings(port);
+        settings.InitKikimrRunConfig()
+                .SetNodeCount(1)
+                .SetUseRealThreads(true)
+                .SetDomainName("Root")
+                .SetMonitoringPortOffset(monPort, true);
+
+        TServer server(settings);
+        server.EnableGRpc(grpcPort);
+        TClient client(settings);
+        client.InitRootScheme();
+
+        TTestActorRuntime& runtime = *server.GetRuntime();
+        runtime.SetLogPriority(NKikimrServices::TICKET_PARSER, NLog::PRI_TRACE);
+
+        TKeepAliveHttpClient httpClient("localhost", monPort);
+
+        PostQuery(httpClient, "CREATE TABLE `/Root/Test` (Key Uint64, Value String, PRIMARY KEY (Key));", "execute-query");
+        PostQuery(httpClient, "INSERT INTO `/Root/Test` (Key, Value) VALUES (1, 'testvalue');", "execute-query");
+        TString response = PostQuery(httpClient, "SELECT * FROM `/Root/Test`;", "execute-data");
+        {
+            NJson::TJsonReaderConfig jsonCfg;
+            NJson::TJsonValue json;
+            NJson::ReadJsonTree(response, &jsonCfg, &json, /* throwOnError = */ true);
+            auto resultSets = json["result"].GetArray();
+            UNIT_ASSERT_EQUAL_C(1, resultSets.size(), response);
+        }
+    }
+
     Y_UNIT_TEST(FloatPointJsonQuery) {
         TPortManager tp;
         ui16 port = tp.GetPort(2134);