Remove TActiveTransaction from datashard_kqp (#1454)

10 files changed, 132 insertions, 154 deletions

diff --git a/ydb/core/tx/datashard/build_kqp_data_tx_out_rs_unit.cpp b/ydb/core/tx/datashard/build_kqp_data_tx_out_rs_unit.cpp
index 04660d8c705..236c04a97aa 100644
--- a/ydb/core/tx/datashard/build_kqp_data_tx_out_rs_unit.cpp
+++ b/ydb/core/tx/datashard/build_kqp_data_tx_out_rs_unit.cpp
@@ -100,7 +100,7 @@ EExecutionStatus TBuildKqpDataTxOutRSUnit::Execute(TOperation::TPtr op, TTransac
         dataTx->SetReadVersion(DataShard.GetReadWriteVersions(tx).ReadVersion);
 
         if (dataTx->GetKqpComputeCtx().HasPersistentChannels()) {
-            auto result = KqpRunTransaction(ctx, op->GetTxId(), kqpLocks, useGenericReadSets, tasksRunner);
+            auto result = KqpRunTransaction(ctx, op->GetTxId(), useGenericReadSets, tasksRunner);
 
             Y_VERIFY_S(!dataTx->GetKqpComputeCtx().HadInconsistentReads(),
                 "Unexpected inconsistent reads in operation " << *op << " when preparing persistent channels");
diff --git a/ydb/core/tx/datashard/datashard__engine_host.cpp b/ydb/core/tx/datashard/datashard__engine_host.cpp
index 8e2d2cbc478..ad98ab4c1c7 100644
--- a/ydb/core/tx/datashard/datashard__engine_host.cpp
+++ b/ydb/core/tx/datashard/datashard__engine_host.cpp
@@ -276,7 +276,7 @@ public:
         return UserDb.GetChangeCollector(tableId);
     }
 
-    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion) {
+    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion) override {
         UserDb.CommitChanges(tableId, lockId, writeVersion);
     }
 
@@ -607,13 +607,6 @@ void TEngineBay::SetIsRepeatableSnapshot() {
     host->SetIsRepeatableSnapshot();
 }
 
-void TEngineBay::CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion) {
-    Y_ABORT_UNLESS(EngineHost);
-
-    auto* host = static_cast<TDataShardEngineHost*>(EngineHost.Get());
-    host->CommitChanges(tableId, lockId, writeVersion);
-}
-
 TVector<IDataShardChangeCollector::TChange> TEngineBay::GetCollectedChanges() const {
     Y_ABORT_UNLESS(EngineHost);
 
diff --git a/ydb/core/tx/datashard/datashard__engine_host.h b/ydb/core/tx/datashard/datashard__engine_host.h
index 3c0808e0f47..7d1c62235cc 100644
--- a/ydb/core/tx/datashard/datashard__engine_host.h
+++ b/ydb/core/tx/datashard/datashard__engine_host.h
@@ -103,8 +103,6 @@ public:
     void SetIsImmediateTx();
     void SetIsRepeatableSnapshot();
 
-    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion);
-
     TVector<IDataShardChangeCollector::TChange> GetCollectedChanges() const;
     void ResetCollectedChanges();
 
diff --git a/ydb/core/tx/datashard/datashard_active_transaction.cpp b/ydb/core/tx/datashard/datashard_active_transaction.cpp
index b54f21d39a8..8d9b8bf2011 100644
--- a/ydb/core/tx/datashard/datashard_active_transaction.cpp
+++ b/ydb/core/tx/datashard/datashard_active_transaction.cpp
@@ -206,6 +206,13 @@ TValidatedDataTx::~TValidatedDataTx() {
     NActors::NMemory::TLabel<MemoryLabelValidatedDataTx>::Sub(TxSize);
 }
 
+TDataShardUserDb& TValidatedDataTx::GetUserDb() {
+    return EngineBay.GetUserDb();
+}
+const TDataShardUserDb& TValidatedDataTx::GetUserDb() const {
+    return EngineBay.GetUserDb();
+}
+
 ui32 TValidatedDataTx::ExtractKeys(bool allowErrors)
 {
     using EResult = NMiniKQL::IEngineFlat::EResult;
diff --git a/ydb/core/tx/datashard/datashard_active_transaction.h b/ydb/core/tx/datashard/datashard_active_transaction.h
index cca5b12b876..dd1870362b7 100644
--- a/ydb/core/tx/datashard/datashard_active_transaction.h
+++ b/ydb/core/tx/datashard/datashard_active_transaction.h
@@ -23,6 +23,7 @@ using NTabletFlatExecutor::TTransactionContext;
 using NTabletFlatExecutor::TTableSnapshotContext;
 
 class TDataShard;
+class TDataShardUserDb;
 class TSysLocks;
 struct TReadSetKey;
 class TActiveTransaction;
@@ -173,6 +174,9 @@ public:
     const NMiniKQL::TEngineHostCounters& GetCounters() { return EngineBay.GetCounters(); }
     void ResetCounters() { EngineBay.ResetCounters(); }
 
+    TDataShardUserDb& GetUserDb();
+    const TDataShardUserDb& GetUserDb() const;
+
     bool CanCancel();
     bool CheckCancelled(ui64 tabletId);
 
@@ -180,10 +184,6 @@ public:
     void SetReadVersion(TRowVersion readVersion) { EngineBay.SetReadVersion(readVersion); }
     void SetVolatileTxId(ui64 txId) { EngineBay.SetVolatileTxId(txId); }
 
-    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion) {
-        EngineBay.CommitChanges(tableId, lockId, writeVersion);
-    }
-
     TVector<IDataShardChangeCollector::TChange> GetCollectedChanges() const { return EngineBay.GetCollectedChanges(); }
     void ResetCollectedChanges() { EngineBay.ResetCollectedChanges(); }
 
diff --git a/ydb/core/tx/datashard/datashard_kqp.cpp b/ydb/core/tx/datashard/datashard_kqp.cpp
index cdd14d30f54..efd84c1ba70 100644
--- a/ydb/core/tx/datashard/datashard_kqp.cpp
+++ b/ydb/core/tx/datashard/datashard_kqp.cpp
@@ -1,5 +1,6 @@
 #include "datashard_kqp.h"
 #include "datashard_impl.h"
+#include "datashard_user_db.h"
 
 #include <ydb/core/kqp/common/kqp.h>
 #include <ydb/core/kqp/runtime/kqp_tasks_runner.h>
@@ -98,8 +99,7 @@ NUdf::EFetchStatus FetchOutput(NDq::IDqOutputChannel* channel, NDq::TDqSerialize
 }
 
 NDq::ERunStatus RunKqpTransactionInternal(const TActorContext& ctx, ui64 txId,
-    const TInputOpData::TInReadSets* inReadSets, const NKikimrDataEvents::TKqpLocks&,
-    bool useGenericReadSets,
+    const TInputOpData::TInReadSets* inReadSets,  bool useGenericReadSets,
     NKqp::TKqpTasksRunner& tasksRunner, bool applyEffects)
 {
     THashMap<ui64, std::pair<ui64, ui32>> inputChannelsMap; // channelId -> (taskId, input index)
@@ -500,18 +500,16 @@ void KqpSetTxLocksKeys(const NKikimrDataEvents::TKqpLocks& locks, const TSysLock
     }
 }
 
-NYql::NDq::ERunStatus KqpRunTransaction(const TActorContext& ctx, ui64 txId,
-    const NKikimrDataEvents::TKqpLocks& kqpLocks, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner)
+NYql::NDq::ERunStatus KqpRunTransaction(const TActorContext& ctx, ui64 txId, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner)
 {
-    return RunKqpTransactionInternal(ctx, txId, /* inReadSets */ nullptr, kqpLocks, useGenericReadSets, tasksRunner, /* applyEffects */ false);
+    return RunKqpTransactionInternal(ctx, txId, /* inReadSets */ nullptr, useGenericReadSets, tasksRunner, /* applyEffects */ false);
 }
 
 THolder<TEvDataShard::TEvProposeTransactionResult> KqpCompleteTransaction(const TActorContext& ctx,
-    ui64 origin, ui64 txId, const TInputOpData::TInReadSets* inReadSets,
-    const NKikimrDataEvents::TKqpLocks& kqpLocks, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner,
+    ui64 origin, ui64 txId, const TInputOpData::TInReadSets* inReadSets, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner,
     const NMiniKQL::TKqpDatashardComputeContext& computeCtx)
 {
-    auto runStatus = RunKqpTransactionInternal(ctx, txId, inReadSets, kqpLocks, useGenericReadSets, tasksRunner, /* applyEffects */ true);
+    auto runStatus = RunKqpTransactionInternal(ctx, txId, inReadSets, useGenericReadSets, tasksRunner, /* applyEffects */ true);
 
     if (computeCtx.HadInconsistentReads()) {
         return nullptr;
@@ -690,54 +688,35 @@ void KqpFillOutReadSets(TOutputOpData::TOutReadSets& outReadSets, const NKikimrD
     }
 }
 
-bool KqpValidateLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks) {
-    auto& kqpLocks = tx->GetDataTx()->GetKqpLocks();
-    bool hasKqpLocks = tx->GetDataTx()->HasKqpLocks();
-
-    if (!hasKqpLocks || !NeedValidateLocks(kqpLocks.GetOp())) {
-        return true;
+std::tuple<bool, TVector<NKikimrDataEvents::TLock>> KqpValidateLocks(ui64 origin, TSysLocks& sysLocks, 
+    const NKikimrDataEvents::TKqpLocks* kqpLocks, bool useGenericReadSets, const TInputOpData::TInReadSets& inReadSets) 
+{
+    if (kqpLocks == nullptr || !NeedValidateLocks(kqpLocks->GetOp())) {
+        return {true, {}};
     }
 
-    bool sendLocks = SendLocks(kqpLocks, origin);
-    YQL_ENSURE(sendLocks == !kqpLocks.GetLocks().empty());
+    bool sendLocks = SendLocks(*kqpLocks, origin);
+    YQL_ENSURE(sendLocks == !kqpLocks->GetLocks().empty());
 
     if (sendLocks) {
-        auto brokenLocks = ValidateLocks(kqpLocks, sysLocks, origin);
+        auto brokenLocks = ValidateLocks(*kqpLocks, sysLocks, origin);
 
         if (!brokenLocks.empty()) {
-            tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
-                NKikimrTxDataShard::TX_KIND_DATA,
-                origin,
-                tx->GetTxId(),
-                NKikimrTxDataShard::TEvProposeTransactionResult::LOCKS_BROKEN);
-
-            auto* protoLocks = tx->Result()->Record.MutableTxLocks();
-            for (auto& brokenLock : brokenLocks) {
-                protoLocks->Add()->Swap(&brokenLock);
-            }
-
-            return false;
+            return {false, std::move(brokenLocks)};
         }
     }
 
-    for (auto& readSet : tx->InReadSets()) {
-        for (auto& data : readSet.second) {
-            if (tx->GetDataTx()->GetUseGenericReadSets()) {
+    for (const auto& readSet : inReadSets) {
+        for (const auto& data : readSet.second) {
+            if (useGenericReadSets) {
                 NKikimrTx::TReadSetData genericData;
                 bool ok = genericData.ParseFromString(data.Body);
-                Y_ABORT_UNLESS(ok, "Failed to parse generic readset from %" PRIu64 " to %" PRIu64 " origin %" PRIu64,
+                Y_ABORT_UNLESS(ok, "Failed to parse generic readset from %" PRIu64 " to %" PRIu64 " tabletId %" PRIu64,
                     readSet.first.first, readSet.first.second, data.Origin);
 
                 if (genericData.GetDecision() != NKikimrTx::TReadSetData::DECISION_COMMIT) {
-                    tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
-                        NKikimrTxDataShard::TX_KIND_DATA,
-                        origin,
-                        tx->GetTxId(),
-                        NKikimrTxDataShard::TEvProposeTransactionResult::LOCKS_BROKEN);
-
                     // Note: we don't know details on what failed at that shard
-
-                    return false;
+                    return {false, {}};
                 }
             } else {
                 NKikimrTxDataShard::TKqpReadset kqpReadset;
@@ -746,63 +725,48 @@ bool KqpValidateLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks)
                 if (kqpReadset.HasValidateLocksResult()) {
                     auto& validateResult = kqpReadset.GetValidateLocksResult();
                     if (!validateResult.GetSuccess()) {
-                        tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
-                            NKikimrTxDataShard::TX_KIND_DATA,
-                            origin,
-                            tx->GetTxId(),
-                            NKikimrTxDataShard::TEvProposeTransactionResult::LOCKS_BROKEN);
-
-                        tx->Result()->Record.MutableTxLocks()->CopyFrom(validateResult.GetBrokenLocks());
-
-                        return false;
+                        TVector<NKikimrDataEvents::TLock> brokenLocks;
+                        brokenLocks.reserve(validateResult.GetBrokenLocks().size());
+                        std::copy(validateResult.GetBrokenLocks().begin(), validateResult.GetBrokenLocks().end(), std::back_inserter(brokenLocks));
+                        return {false, std::move(brokenLocks)};
                     }
                 }
             }
         }
     }
 
-    return true;
+    return {true, {}};
 }
 
-bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks) {
-    auto& kqpLocks = tx->GetDataTx()->GetKqpLocks();
-
-    if (!tx->GetDataTx()->HasKqpLocks() || !NeedValidateLocks(kqpLocks.GetOp())) {
-        return true;
+std::tuple<bool, TVector<NKikimrDataEvents::TLock>> KqpValidateVolatileTx(ui64 origin, TSysLocks& sysLocks, 
+    const NKikimrDataEvents::TKqpLocks* kqpLocks, bool useGenericReadSets, ui64 txId, const TVector<NKikimrTx::TEvReadSet>& delayedInReadSets, 
+    TInputOpData::TAwaitingDecisions& awaitingDecisions, TOutputOpData::TOutReadSets& outReadSets) 
+{
+    if (kqpLocks == nullptr || !NeedValidateLocks(kqpLocks->GetOp())) {
+        return {true, {}};
     }
 
     // Volatile transactions cannot work with non-generic readsets
-    YQL_ENSURE(tx->GetDataTx()->GetUseGenericReadSets());
+    YQL_ENSURE(useGenericReadSets);
 
     // We may have some stale data since before the restart
     // We expect all stale data to be cleared on restarts
-    Y_ABORT_UNLESS(tx->OutReadSets().empty());
-    Y_ABORT_UNLESS(tx->AwaitingDecisions().empty());
+    Y_ABORT_UNLESS(outReadSets.empty());
+    Y_ABORT_UNLESS(awaitingDecisions.empty());
 
     // Note: usually all shards send locks, since they either have side effects or need to validate locks
     // However it is technically possible to have pure-read shards, that don't contribute to the final decision
-    bool sendLocks = SendLocks(kqpLocks, origin);
+    bool sendLocks = SendLocks(*kqpLocks, origin);
     if (sendLocks) {
         // Note: it is possible to have no locks
-        auto brokenLocks = ValidateLocks(kqpLocks, sysLocks, origin);
+        auto brokenLocks = ValidateLocks(*kqpLocks, sysLocks, origin);
 
         if (!brokenLocks.empty()) {
-            tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
-                NKikimrTxDataShard::TX_KIND_DATA,
-                origin,
-                tx->GetTxId(),
-                NKikimrTxDataShard::TEvProposeTransactionResult::LOCKS_BROKEN);
-
-            auto* protoLocks = tx->Result()->Record.MutableTxLocks();
-            for (auto& brokenLock : brokenLocks) {
-                protoLocks->Add()->Swap(&brokenLock);
-            }
-
-            return false;
+            return {false, std::move(brokenLocks)};
         }
 
         // We need to form decision readsets for all other participants
-        for (ui64 dstTabletId : kqpLocks.GetReceivingShards()) {
+        for (ui64 dstTabletId : kqpLocks->GetReceivingShards()) {
             if (dstTabletId == origin) {
                 // Don't send readsets to ourselves
                 continue;
@@ -819,15 +783,15 @@ bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLo
             bool ok = data.SerializeToString(&bodyStr);
             Y_ABORT_UNLESS(ok, "Failed to serialize readset from %" PRIu64 " to %" PRIu64, key.first, key.second);
 
-            tx->OutReadSets()[key] = std::move(bodyStr);
+            outReadSets[key] = std::move(bodyStr);
         }
     }
 
-    bool receiveLocks = ReceiveLocks(kqpLocks, origin);
+    bool receiveLocks = ReceiveLocks(*kqpLocks, origin);
     if (receiveLocks) {
         // Note: usually only shards with side-effects receive locks, since they
         //       need the final outcome to decide whether to commit or abort.
-        for (ui64 srcTabletId : kqpLocks.GetSendingShards()) {
+        for (ui64 srcTabletId : kqpLocks->GetSendingShards()) {
             if (srcTabletId == origin) {
                 // Don't await decision from ourselves
                 continue;
@@ -836,20 +800,20 @@ bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLo
             LOG_TRACE_S(*TlsActivationContext, NKikimrServices::TX_DATASHARD, "Will wait for volatile decision from "
                 << srcTabletId << " to " << origin);
 
-            tx->AwaitingDecisions().insert(srcTabletId);
+            awaitingDecisions.insert(srcTabletId);
         }
 
         bool aborted = false;
 
-        for (auto& record : tx->DelayedInReadSets()) {
+        for (const auto& record : delayedInReadSets) {
             ui64 srcTabletId = record.GetTabletSource();
             ui64 dstTabletId = record.GetTabletDest();
             if (dstTabletId != origin) {
                 LOG_WARN_S(*TlsActivationContext, NKikimrServices::TX_DATASHARD, "Ignoring unexpected readset from "
-                    << srcTabletId << " to " << dstTabletId << " for txId# " << tx->GetTxId() << " at tablet " << origin);
+                    << srcTabletId << " to " << dstTabletId << " for txId# " << txId << " at tablet " << origin);
                 continue;
             }
-            if (!tx->AwaitingDecisions().contains(srcTabletId)) {
+            if (!awaitingDecisions.contains(srcTabletId)) {
                 continue;
             }
 
@@ -859,7 +823,7 @@ bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLo
 
                 // No readset data: participant aborted the transaction
                 LOG_TRACE_S(*TlsActivationContext, NKikimrServices::TX_DATASHARD, "Processed readset without data from"
-                    << srcTabletId << " to " << dstTabletId << " will abort txId# " << tx->GetTxId());
+                    << srcTabletId << " to " << dstTabletId << " will abort txId# " << txId);
                 aborted = true;
                 break;
             }
@@ -872,38 +836,30 @@ bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLo
                 // Explicit decision that is not a commit, need to abort
                 LOG_TRACE_S(*TlsActivationContext, NKikimrServices::TX_DATASHARD, "Processed decision "
                     << ui32(data.GetDecision()) << " from " << srcTabletId << " to " << dstTabletId
-                    << " for txId# " << tx->GetTxId());
+                    << " for txId# " << txId);
                 aborted = true;
                 break;
             }
 
             LOG_TRACE_S(*TlsActivationContext, NKikimrServices::TX_DATASHARD, "Processed commit decision from "
-                << srcTabletId << " to " << dstTabletId << " for txId# " << tx->GetTxId());
-            tx->AwaitingDecisions().erase(srcTabletId);
+                << srcTabletId << " to " << dstTabletId << " for txId# " << txId);
+            awaitingDecisions.erase(srcTabletId);
         }
 
         if (aborted) {
-            tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
-                NKikimrTxDataShard::TX_KIND_DATA,
-                origin,
-                tx->GetTxId(),
-                NKikimrTxDataShard::TEvProposeTransactionResult::ABORTED);
-
-            return false;
+            return {false, {}};
         }
     }
 
-    return true;
+    return {true, {}};
 }
 
-void KqpEraseLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks) {
-    auto& kqpLocks = tx->GetDataTx()->GetKqpLocks();
-
-    if (!tx->GetDataTx()->HasKqpLocks() || !NeedEraseLocks(kqpLocks.GetOp())) {
+void KqpEraseLocks(ui64 origin, const NKikimrDataEvents::TKqpLocks* kqpLocks, TSysLocks& sysLocks) {
+    if (kqpLocks == nullptr || !NeedEraseLocks(kqpLocks->GetOp())) {
         return;
     }
 
-    for (auto& lockProto : kqpLocks.GetLocks()) {
+    for (const auto& lockProto : kqpLocks->GetLocks()) {
         if (lockProto.GetDataShard() != origin) {
             continue;
         }
@@ -915,18 +871,14 @@ void KqpEraseLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks) {
     }
 }
 
-void KqpCommitLocks(ui64 origin, TActiveTransaction* tx, const TRowVersion& writeVersion, TDataShard& dataShard) {
-
-    if (!tx->GetDataTx()->HasKqpLocks()) {
+void KqpCommitLocks(ui64 origin, const NKikimrDataEvents::TKqpLocks* kqpLocks, TSysLocks& sysLocks, const TRowVersion& writeVersion, IDataShardUserDb& userDb) {
+    if (kqpLocks == nullptr) {
         return;
     }
 
-    auto& kqpLocks = tx->GetDataTx()->GetKqpLocks();
-    TSysLocks& sysLocks = dataShard.SysLocksTable();
-
-    if (NeedCommitLocks(kqpLocks.GetOp())) {
+    if (NeedCommitLocks(kqpLocks->GetOp())) {
         // We assume locks have been validated earlier
-        for (auto& lockProto : kqpLocks.GetLocks()) {
+        for (const auto& lockProto : kqpLocks->GetLocks()) {
             if (lockProto.GetDataShard() != origin) {
                 continue;
             }
@@ -939,10 +891,10 @@ void KqpCommitLocks(ui64 origin, TActiveTransaction* tx, const TRowVersion& writ
             TTableId tableId(lockProto.GetSchemeShard(), lockProto.GetPathId());
             auto txId = lockProto.GetLockId();
 
-            tx->GetDataTx()->CommitChanges(tableId, txId, writeVersion);
+            userDb.CommitChanges(tableId, txId, writeVersion);
         }
     } else {
-        KqpEraseLocks(origin, tx, sysLocks);
+        KqpEraseLocks(origin, kqpLocks, sysLocks);
     }
 }
 
diff --git a/ydb/core/tx/datashard/datashard_kqp.h b/ydb/core/tx/datashard/datashard_kqp.h
index 54b9c474695..12c230e92f1 100644
--- a/ydb/core/tx/datashard/datashard_kqp.h
+++ b/ydb/core/tx/datashard/datashard_kqp.h
@@ -1,7 +1,11 @@
 #pragma once
 
 #include "datashard.h"
-#include "datashard_active_transaction.h"
+
+#include "operation.h"
+#include "key_validator.h"
+#include "datashard_locks.h"
+#include "datashard_user_db.h"
 
 #include <ydb/core/engine/minikql/minikql_engine_host.h>
 #include <ydb/core/kqp/runtime/kqp_tasks_runner.h>
@@ -20,12 +24,10 @@ void KqpSetTxKeys(ui64 tabletId, ui64 taskId, const TUserTable* tableInfo,
 
 void KqpSetTxLocksKeys(const NKikimrDataEvents::TKqpLocks& locks, const TSysLocks& sysLocks, TKeyValidator& keyValidator);
 
-NYql::NDq::ERunStatus KqpRunTransaction(const TActorContext& ctx, ui64 txId,
-    const NKikimrDataEvents::TKqpLocks& kqpLocks, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner);
+NYql::NDq::ERunStatus KqpRunTransaction(const TActorContext& ctx, ui64 txId, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner);
 
 THolder<TEvDataShard::TEvProposeTransactionResult> KqpCompleteTransaction(const TActorContext& ctx,
-    ui64 origin, ui64 txId, const TInputOpData::TInReadSets* inReadSets,
-    const NKikimrDataEvents::TKqpLocks& kqpLocks, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner,
+    ui64 tabletId, ui64 txId, const TInputOpData::TInReadSets* inReadSets, bool useGenericReadSets, NKqp::TKqpTasksRunner& tasksRunner,
     const NMiniKQL::TKqpDatashardComputeContext& computeCtx);
 
 void KqpFillOutReadSets(TOutputOpData::TOutReadSets& outReadSets, const NKikimrDataEvents::TKqpLocks& kqpLocks, 
@@ -35,11 +37,14 @@ void KqpFillOutReadSets(TOutputOpData::TOutReadSets& outReadSets, const NKikimrD
 void KqpPrepareInReadsets(TInputOpData::TInReadSets& inReadSets,
     const NKikimrDataEvents::TKqpLocks& kqpLocks, const NKqp::TKqpTasksRunner& tasksRunner, ui64 tabletId);
 
-bool KqpValidateLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks);
-bool KqpValidateVolatileTx(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks);
+std::tuple<bool, TVector<NKikimrDataEvents::TLock>> KqpValidateLocks(ui64 tabletId, TSysLocks& sysLocks, 
+    const NKikimrDataEvents::TKqpLocks* kqpLocks, bool useGenericReadSets, const TInputOpData::TInReadSets& inReadSets);
+std::tuple<bool, TVector<NKikimrDataEvents::TLock>> KqpValidateVolatileTx(ui64 tabletId, TSysLocks& sysLocks, 
+    const NKikimrDataEvents::TKqpLocks* kqpLocks, bool useGenericReadSets, ui64 txId, const TVector<NKikimrTx::TEvReadSet>& delayedInReadSets, 
+    TInputOpData::TAwaitingDecisions& awaitingDecisions, TOutputOpData::TOutReadSets& outReadSets);
 
-void KqpEraseLocks(ui64 origin, TActiveTransaction* tx, TSysLocks& sysLocks);
-void KqpCommitLocks(ui64 origin, TActiveTransaction* tx, const TRowVersion& writeVersion, TDataShard& dataShard);
+void KqpEraseLocks(ui64 tabletId, const NKikimrDataEvents::TKqpLocks* kqpLocks, TSysLocks& sysLocks);
+void KqpCommitLocks(ui64 tabletId, const NKikimrDataEvents::TKqpLocks* kqpLocks, TSysLocks& sysLocks, const TRowVersion& writeVersion, IDataShardUserDb& userDb);
 
 void KqpUpdateDataShardStatCounters(TDataShard& dataShard, const NMiniKQL::TEngineHostCounters& counters);
 
diff --git a/ydb/core/tx/datashard/datashard_user_db.h b/ydb/core/tx/datashard/datashard_user_db.h
index 2bf1f72fb90..04a856d4757 100644
--- a/ydb/core/tx/datashard/datashard_user_db.h
+++ b/ydb/core/tx/datashard/datashard_user_db.h
@@ -32,7 +32,12 @@ public:
 
     virtual void EraseRow(
             const TTableId& tableId,
-            const TArrayRef<const TRawTypeValue> key) = 0;            
+            const TArrayRef<const TRawTypeValue> key) = 0;
+
+    virtual void CommitChanges(
+            const TTableId& tableId,
+            ui64 lockId,
+            const TRowVersion& writeVersion) = 0;
 };
 
 class IDataShardConflictChecker {
@@ -89,6 +94,11 @@ public:
             const TTableId& tableId,
             const TArrayRef<const TRawTypeValue> key) override;
 
+    void CommitChanges(
+            const TTableId& tableId, 
+            ui64 lockId, 
+            const TRowVersion& writeVersion) override;
+
 //IDataShardChangeGroupProvider
 public:  
     std::optional<ui64> GetCurrentChangeGroup() const override;
@@ -110,7 +120,6 @@ public:
     void ResetCollectedChanges();
 
 public:
-    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion);
     bool NeedToReadBeforeWrite(const TTableId& tableId);
     void AddCommitTxId(const TTableId& tableId, ui64 txId, const TRowVersion& commitVersion);
     ui64 GetWriteTxId(const TTableId& tableId);
diff --git a/ydb/core/tx/datashard/datashard_write_operation.h b/ydb/core/tx/datashard/datashard_write_operation.h
index fa0bdf2fc95..b14d829244c 100644
--- a/ydb/core/tx/datashard/datashard_write_operation.h
+++ b/ydb/core/tx/datashard/datashard_write_operation.h
@@ -103,10 +103,6 @@ public:
         UserDb.SetVolatileTxId(txId);
     }
 
-    void CommitChanges(const TTableId& tableId, ui64 lockId, const TRowVersion& writeVersion) {
-        UserDb.CommitChanges(tableId, lockId, writeVersion);
-    }
-
     TVector<IDataShardChangeCollector::TChange> GetCollectedChanges() const {
         return UserDb.GetCollectedChanges();
     }
diff --git a/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp b/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
index cd331d8f7f3..3808fc2d4d4 100644
--- a/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
+++ b/ydb/core/tx/datashard/execute_kqp_data_tx_unit.cpp
@@ -4,6 +4,7 @@
 #include "execution_unit_ctors.h"
 #include "setup_sys_locks.h"
 #include "datashard_locks_db.h"
+#include "datashard_user_db.h"
 #include "probes.h"
 
 #include <ydb/core/engine/minikql/minikql_engine_host.h>
@@ -126,9 +127,14 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
     }
 
     try {
-        auto& kqpLocks = dataTx->GetKqpLocks();
+        const ui64 txId = tx->GetTxId();
+        const auto* kqpLocks = tx->GetDataTx()->HasKqpLocks() ? &tx->GetDataTx()->GetKqpLocks() : nullptr;
+        const auto& inReadSets = op->InReadSets();
+        auto& awaitingDecisions = tx->AwaitingDecisions();
+        auto& outReadSets = tx->OutReadSets();
         bool useGenericReadSets = dataTx->GetUseGenericReadSets();
         auto& tasksRunner = dataTx->GetKqpTasksRunner();
+        TSysLocks& sysLocks = DataShard.SysLocksTable();
 
         ui64 consumedMemory = dataTx->GetTxSize() + tasksRunner.GetAllocatedMemory();
         if (MaybeRequestMoreTxMemory(consumedMemory, txc)) {
@@ -177,18 +183,30 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
             // We need to clear OutReadSets and AwaitingDecisions for
             // volatile transactions, except when we commit them.
             if (!keepOutReadSets) {
-                tx->OutReadSets().clear();
-                tx->AwaitingDecisions().clear();
+                outReadSets.clear();
+                awaitingDecisions.clear();
             }
         };
 
-        const bool validated = op->HasVolatilePrepareFlag()
-            ? KqpValidateVolatileTx(tabletId, tx, DataShard.SysLocksTable())
-            : KqpValidateLocks(tabletId, tx, DataShard.SysLocksTable());
+        auto [validated, brokenLocks] = op->HasVolatilePrepareFlag()
+            ? KqpValidateVolatileTx(tabletId, sysLocks, kqpLocks, useGenericReadSets, 
+                txId, tx->DelayedInReadSets(), awaitingDecisions, outReadSets)
+            : KqpValidateLocks(tabletId, sysLocks, kqpLocks, useGenericReadSets, inReadSets);
 
         if (!validated) {
-            KqpEraseLocks(tabletId, tx, DataShard.SysLocksTable());
-            DataShard.SysLocksTable().ApplyLocks();
+            tx->Result() = MakeHolder<TEvDataShard::TEvProposeTransactionResult>(
+                NKikimrTxDataShard::TX_KIND_DATA,
+                tabletId,
+                txId,
+                NKikimrTxDataShard::TEvProposeTransactionResult::LOCKS_BROKEN
+            );
+
+            for (auto& brokenLock : brokenLocks) {
+                tx->Result()->Record.MutableTxLocks()->Add()->Swap(&brokenLock);
+            }
+
+            KqpEraseLocks(tabletId, kqpLocks, sysLocks);
+            sysLocks.ApplyLocks();
             DataShard.SubscribeNewLocks(ctx);
             if (locksDb.HasChanges()) {
                 op->SetWaitCompletionFlag(true);
@@ -203,10 +221,10 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
         req.MemoryPool = NKqp::NRm::EKqpMemoryPool::DataQuery;
         req.Memory = txc.GetMemoryLimit();
         ui64 taskId = dataTx->GetFirstKqpTaskId();
-        NKqp::GetKqpResourceManager()->NotifyExternalResourcesAllocated(tx->GetTxId(), taskId, req);
+        NKqp::GetKqpResourceManager()->NotifyExternalResourcesAllocated(txId, taskId, req);
 
         Y_DEFER {
-            NKqp::GetKqpResourceManager()->NotifyExternalResourcesFreed(tx->GetTxId(), taskId);
+            NKqp::GetKqpResourceManager()->NotifyExternalResourcesFreed(txId, taskId);
         };
 
         LOG_T("Operation " << *op << " (execute_kqp_data_tx) at " << tabletId
@@ -220,17 +238,17 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
         dataTx->SetWriteVersion(writeVersion);
 
         if (op->HasVolatilePrepareFlag()) {
-            dataTx->SetVolatileTxId(tx->GetTxId());
+            dataTx->SetVolatileTxId(txId);
         }
 
         LWTRACK(ProposeTransactionKqpDataExecute, op->Orbit);
 
-        KqpCommitLocks(tabletId, tx, writeVersion, DataShard);
+        KqpCommitLocks(tabletId, kqpLocks, sysLocks, writeVersion, tx->GetDataTx()->GetUserDb());
 
         auto& computeCtx = tx->GetDataTx()->GetKqpComputeCtx();
 
         auto result = KqpCompleteTransaction(ctx, tabletId, op->GetTxId(),
-            op->HasKqpAttachedRSFlag() ? nullptr : &op->InReadSets(), kqpLocks, useGenericReadSets, tasksRunner, computeCtx);
+            op->HasKqpAttachedRSFlag() ? nullptr : &op->InReadSets(), useGenericReadSets, tasksRunner, computeCtx);
 
         if (!result && computeCtx.HadInconsistentReads()) {
             LOG_T("Operation " << *op << " (execute_kqp_data_tx) at " << tabletId
@@ -319,16 +337,16 @@ EExecutionStatus TExecuteKqpDataTxUnit::Execute(TOperation::TPtr op, TTransactio
         op->SetKqpAttachedRSFlag();
 
         if (dataTx->GetCounters().InvisibleRowSkips && op->LockTxId()) {
-            DataShard.SysLocksTable().BreakSetLocks();
+            sysLocks.BreakSetLocks();
         }
 
         // Note: any transaction (e.g. immediate or non-volatile) may decide to commit as volatile due to dependencies
         // Such transactions would have no participants and become immediately committed
         auto commitTxIds = dataTx->GetVolatileCommitTxIds();
         if (commitTxIds) {
-            TVector<ui64> participants(tx->AwaitingDecisions().begin(), tx->AwaitingDecisions().end());
+            TVector<ui64> participants(awaitingDecisions.begin(), awaitingDecisions.end());
             DataShard.GetVolatileTxManager().PersistAddVolatileTx(
-                tx->GetTxId(),
+                txId,
                 writeVersion,
                 commitTxIds,
                 dataTx->GetVolatileDependencies(),