fix

Author: nikvas0

ydb/core/kqp/common/kqp_tx_manager.cpp

@@ -189,18 +189,16 @@ class TKqpTransactionManager : public IKqpTransactionManager {
         return !HasOlapTable();
     }
 
-    const std::optional<NKikimrDataEvents::TMvccSnapshot>& GetSnapshot() const override {
-        return Snapshot;
+    bool HasSnapshot() const override {
+        return ValidSnapshot;
     }
 
-    void SetSnapshot(ui64 step, ui64 txId) override {
-        Snapshot.emplace();
-        Snapshot->SetStep(step);
-        Snapshot->SetTxId(txId);
+    void SetHasSnapshot(bool hasSnapshot) override {
+        ValidSnapshot = hasSnapshot;
     }
 
     bool BrokenLocks() const override {
-        return LocksIssue.has_value() && !(GetSnapshot() && IsReadOnly());
+        return LocksIssue.has_value() && !(HasSnapshot() && IsReadOnly());
     }
 
     const std::optional<NYql::TIssue>& GetLockIssue() const override {
@@ -407,8 +405,8 @@ class TKqpTransactionManager : public IKqpTransactionManager {
     std::unordered_set<TString> TablePathes;
 
     bool ReadOnly = true;
+    bool ValidSnapshot = false;
     bool HasOlapTableShard = false;
-    std::optional<NKikimrDataEvents::TMvccSnapshot> Snapshot;
     std::optional<NYql::TIssue> LocksIssue;
 
     THashSet<ui64> SendingShards;

ydb/core/kqp/common/kqp_tx_manager.h

@@ -61,8 +61,8 @@ class IKqpTransactionManager {
 
     virtual bool IsVolatile() const = 0;
 
-    virtual const std::optional<NKikimrDataEvents::TMvccSnapshot>& GetSnapshot() const = 0;
-    virtual void SetSnapshot(ui64 step, ui64 txId) = 0;
+    virtual bool HasSnapshot() const = 0;
+    virtual void SetHasSnapshot(bool hasSnapshot) = 0;
 
     virtual bool BrokenLocks() const = 0;
     virtual const std::optional<NYql::TIssue>& GetLockIssue() const = 0;

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -282,6 +282,10 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
             }
         }
 
+        if (TxManager) {
+            TxManager->SetHasSnapshot(GetSnapshot().IsValid());
+        }
+
         if (!BufferActorId || (ReadOnlyTx && Request.LocksOp != ELocksOp::Rollback)) {
             Become(&TKqpDataExecuter::FinalizeState);
             MakeResponseAndPassAway();
@@ -2237,8 +2241,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
             ExecuterStateSpan = NWilson::TSpan(TWilsonKqp::DataExecuterAcquireSnapshot, ExecuterSpan.GetTraceId(), "WaitForSnapshot");
 
             return;
-        } else if (TxManager && GetSnapshot().IsValid()) {
-            TxManager->SetSnapshot(GetSnapshot().Step, GetSnapshot().TxId);
         }
 
         ContinueExecute();
@@ -2274,9 +2276,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
 
         SetSnapshot(record.GetSnapshotStep(), record.GetSnapshotTxId());
         ImmediateTx = true;
-        if (TxManager) {
-            TxManager->SetSnapshot(GetSnapshot().Step, GetSnapshot().TxId);
-        }
 
         ContinueExecute();
     }

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -170,7 +170,7 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
         const bool inconsistentTx,
         const NMiniKQL::TTypeEnvironment& typeEnv,
         std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> alloc,
-        const NKikimrDataEvents::TMvccSnapshot& mvccSnapshot,
+        const std::optional<NKikimrDataEvents::TMvccSnapshot>& mvccSnapshot,
         const IKqpTransactionManagerPtr& txManager,
         const TActorId sessionActorId,
         TIntrusivePtr<TKqpCounters> counters,
@@ -815,7 +815,9 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
             FillEvWritePrepare(evWrite.get(), shardId, *TxId, TxManager);
         } else if (!InconsistentTx) {
             evWrite->SetLockId(LockTxId, LockNodeId);
-            *evWrite->Record.MutableMvccSnapshot() = MvccSnapshot;
+            if (MvccSnapshot) {
+                *evWrite->Record.MutableMvccSnapshot() = *MvccSnapshot;
+            }
         }
 
         const auto serializationResult = ShardedWriteController->SerializeMessageToPayload(shardId, *evWrite);
@@ -957,7 +959,7 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
     const NMiniKQL::TTypeEnvironment& TypeEnv;
     std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
 
-    NKikimrDataEvents::TMvccSnapshot MvccSnapshot;
+    std::optional<NKikimrDataEvents::TMvccSnapshot> MvccSnapshot;
 
     const TTableId TableId;
     const TString TablePath;
@@ -1024,7 +1026,9 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
             Settings.GetInconsistentTx(),
             TypeEnv,
             Alloc,
-            Settings.GetMvccSnapshot(),
+            Settings.GetIsOlap()
+                ? std::optional<NKikimrDataEvents::TMvccSnapshot>{Settings.GetMvccSnapshot()}
+                : std::optional<NKikimrDataEvents::TMvccSnapshot>{},
             nullptr,
             TActorId{},
             Counters,
@@ -1323,14 +1327,10 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
                 AFL_ENSURE(LockTxId == settings.TransactionSettings.LockTxId);
                 AFL_ENSURE(LockNodeId == settings.TransactionSettings.LockNodeId);
                 AFL_ENSURE(InconsistentTx == settings.TransactionSettings.InconsistentTx);
-                AFL_ENSURE(MvccSnapshot.GetStep() == settings.TransactionSettings.MvccSnapshot.GetStep());
-                AFL_ENSURE(MvccSnapshot.GetTxId() == settings.TransactionSettings.MvccSnapshot.GetTxId());
             } else {
                 LockTxId = settings.TransactionSettings.LockTxId;
                 LockNodeId = settings.TransactionSettings.LockNodeId;
                 InconsistentTx = settings.TransactionSettings.InconsistentTx;
-                MvccSnapshot.SetStep(settings.TransactionSettings.MvccSnapshot.GetStep());
-                MvccSnapshot.SetTxId(settings.TransactionSettings.MvccSnapshot.GetTxId());
             }
 
             auto& writeInfo = WriteInfos[settings.TableId];
@@ -1344,7 +1344,7 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
                     InconsistentTx,
                     TypeEnv,
                     Alloc,
-                    MvccSnapshot,
+                    std::nullopt,
                     TxManager,
                     SessionActorId,
                     Counters,
@@ -2023,7 +2023,6 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
 
     std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
     NMiniKQL::TTypeEnvironment TypeEnv;
-    NKikimrDataEvents::TMvccSnapshot MvccSnapshot;
 
     struct TWriteInfo {
         TKqpTableWriteActor* WriteTableActor = nullptr;
@@ -2150,7 +2149,6 @@ class TKqpForwardWriteActor : public TActorBootstrapped<TKqpForwardWriteActor>,
                     .LockTxId = Settings.GetLockTxId(),
                     .LockNodeId = Settings.GetLockNodeId(),
                     .InconsistentTx = Settings.GetInconsistentTx(),
-                    .MvccSnapshot = Settings.GetMvccSnapshot(),
                 },
                 .Priority = Settings.GetPriority(),
             };