Merge b486489 into 1c39afc

Author: UgnineSirdis

ydb/core/grpc_services/query/rpc_execute_script.cpp

@@ -59,10 +59,6 @@ std::tuple<Ydb::StatusIds::StatusCode, NYql::TIssues> FillKqpRequest(
     kqpRequest.MutableRequest()->SetType(NKikimrKqp::QUERY_TYPE_SQL_GENERIC_SCRIPT);
     kqpRequest.MutableRequest()->SetKeepSession(false);
 
-    // TODO: Avoid explicit tx_control for script queries.
-    kqpRequest.MutableRequest()->MutableTxControl()->mutable_begin_tx()->mutable_serializable_read_write();
-    kqpRequest.MutableRequest()->MutableTxControl()->set_commit_tx(true);
-
     kqpRequest.MutableRequest()->SetCancelAfterMs(GetDuration(req.operation_params().cancel_after()).MilliSeconds());
     kqpRequest.MutableRequest()->SetTimeoutMs(GetDuration(req.operation_params().operation_timeout()).MilliSeconds());
 

ydb/core/kqp/session_actor/kqp_query_state.cpp

@@ -306,4 +306,21 @@ bool TKqpQueryState::HasErrors(const NSchemeCache::TSchemeCacheNavigate& respons
     return true;
 }
 
+bool TKqpQueryState::HasImpliedAutostartTransactions() const {
+    if (!HasTxControl()
+        && (RequestEv->GetAction() == NKikimrKqp::QUERY_ACTION_EXECUTE
+            || RequestEv->GetAction() == NKikimrKqp::QUERY_ACTION_EXECUTE_PREPARED)
+        && (RequestEv->GetType() == NKikimrKqp::QUERY_TYPE_SQL_GENERIC_QUERY
+            || RequestEv->GetType() == NKikimrKqp::QUERY_TYPE_SQL_GENERIC_SCRIPT
+            || RequestEv->GetType() == NKikimrKqp::QUERY_TYPE_SQL_GENERIC_CONCURRENT_QUERY))
+    {
+        for (const auto& transactionPtr : PreparedQuery->GetTransactions()) {
+            if (transactionPtr->GetType() == NKqpProto::TKqpPhyTx::TYPE_GENERIC) { // data transaction
+                return true;
+            }
+        }
+    }
+    return false;
+}
+
 }

ydb/core/kqp/session_actor/kqp_query_state.h

@@ -362,6 +362,8 @@ class TKqpQueryState : public TNonCopyable {
         return RequestEv->HasTxControl();
     }
 
+    bool HasImpliedAutostartTransactions() const;
+
     const ::Ydb::Table::TransactionControl& GetTxControl() const {
         return RequestEv->GetTxControl();
     }

ydb/core/kqp/session_actor/kqp_session_actor.cpp

@@ -636,9 +636,21 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         Counters->ReportBeginTransaction(Settings.DbCounters, Transactions.EvictedTx, Transactions.Size(), Transactions.ToBeAbortedSize());
     }
 
+    static const Ydb::Table::TransactionControl& GetAutoTransactionControl() {
+        auto create = []() -> Ydb::Table::TransactionControl {
+            Ydb::Table::TransactionControl control;
+            control.mutable_begin_tx()->mutable_serializable_read_write();
+            control.set_commit_tx(true);
+            return control;
+        };
+        static const Ydb::Table::TransactionControl control = create();
+        return control;
+    }
+
     bool PrepareQueryTransaction() {
-        if (QueryState->HasTxControl()) {
-            const auto& txControl = QueryState->GetTxControl();
+        bool autoTransaction = false;
+        if (QueryState->HasTxControl() || (autoTransaction = QueryState->HasImpliedAutostartTransactions())) {
+            const auto& txControl = autoTransaction ? GetAutoTransactionControl() : QueryState->GetTxControl();
 
             QueryState->Commit = txControl.commit_tx();
             switch (txControl.tx_selector_case()) {
@@ -941,7 +953,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
                 case NKqpProto::TKqpPhyTx::TYPE_SCHEME:
                     YQL_ENSURE(tx->StagesSize() == 0);
 
-                    if (QueryState->TxCtx->EffectiveIsolationLevel != NKikimrKqp::ISOLATION_LEVEL_UNDEFINED) {
+                    if (QueryState->HasTxControl() && QueryState->TxCtx->EffectiveIsolationLevel != NKikimrKqp::ISOLATION_LEVEL_UNDEFINED) {
                         ReplyQueryError(Ydb::StatusIds::PRECONDITION_FAILED,
                             "Scheme operations cannot be executed inside transaction");
                         return true;

ydb/core/kqp/ut/service/kqp_qs_queries_ut.cpp

@@ -1463,6 +1463,45 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
     }
 
+    Y_UNIT_TEST(DdlExecuteScript) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->SetEnablePreparedDdl(true);
+        auto setting = NKikimrKqp::TKqpSetting();
+        auto serverSettings = TKikimrSettings()
+            .SetAppConfig(appConfig)
+            .SetKqpSettings({setting})
+            .SetEnableScriptExecutionOperations(true);
+
+        TKikimrRunner kikimr(serverSettings);
+        auto db = kikimr.GetQueryClient();
+
+        const TString sql = R"sql(
+            CREATE TABLE TestDdlExecuteScript (
+                Key Uint64,
+                Value String,
+                PRIMARY KEY (Key)
+            );
+        )sql";
+
+        auto scriptExecutionOperation = db.ExecuteScript(sql).ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(scriptExecutionOperation.Status().GetStatus(), EStatus::SUCCESS, scriptExecutionOperation.Status().GetIssues().ToString());
+        UNIT_ASSERT(scriptExecutionOperation.Metadata().ExecutionId);
+
+        NYdb::NOperation::TOperationClient client(kikimr.GetDriver());
+        TMaybe<NYdb::NQuery::TScriptExecutionOperation> readyOp;
+        while (true) {
+            auto op = client.Get<NYdb::NQuery::TScriptExecutionOperation>(scriptExecutionOperation.Id()).GetValueSync();
+            if (op.Ready()) {
+                readyOp = std::move(op);
+                break;
+            }
+            UNIT_ASSERT_C(op.Status().IsSuccess(), TStringBuilder() << op.Status().GetStatus() << ":" << op.Status().GetIssues().ToString());
+            Sleep(TDuration::MilliSeconds(10));
+        }
+        UNIT_ASSERT_C(readyOp->Status().IsSuccess(), readyOp->Status().GetIssues().ToString());
+        UNIT_ASSERT_EQUAL_C(readyOp->Metadata().ExecStatus, EExecStatus::Completed, readyOp->Status().GetIssues().ToString());
+    }
+
     Y_UNIT_TEST(DdlMixedDml) {
         NKikimrConfig::TAppConfig appConfig;
         appConfig.MutableTableServiceConfig()->SetEnablePreparedDdl(true);
@@ -1495,7 +1534,7 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {
             UPSERT INTO KeyValue (Key, Value) VALUES (3, "Three");
             SELECT * FROM KeyValue;
         )", TTxControl::NoTx()).ExtractValueSync();
-        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
     }
 
     Y_UNIT_TEST(Tcl) {

ydb/core/kqp/ut/service/kqp_qs_scripts_ut.cpp

@@ -203,7 +203,7 @@ Y_UNIT_TEST_SUITE(KqpQueryServiceScripts) {
     }
 
 
-    void ExecuteScriptWithStatsMode (Ydb::Query::StatsMode statsMode) {
+    void ExecuteScriptWithStatsMode(Ydb::Query::StatsMode statsMode) {
         auto kikimr = DefaultKikimrRunner();
         auto db = kikimr.GetQueryClient();