Merge 9334977 into 02d2031

Author: nikvas0

ydb/core/kqp/common/kqp_tx.cpp

@@ -169,6 +169,10 @@ bool NeedSnapshot(const TKqpTransactionContext& txCtx, const NYql::TKikimrConfig
     if (!commitTx)
         return true;
 
+    if (HasOlapTableWriteInTx(physicalQuery) || HasOlapTableReadInTx(physicalQuery)) {
+        return true;
+    }
+
     size_t readPhases = 0;
     bool hasEffects = false;
     bool hasSourceRead = false;

ydb/core/kqp/common/kqp_tx_manager.cpp

@@ -189,16 +189,18 @@ class TKqpTransactionManager : public IKqpTransactionManager {
         return !HasOlapTable();
     }
 
-    bool HasSnapshot() const override {
-        return ValidSnapshot;
+    const std::optional<NKikimrDataEvents::TMvccSnapshot>& GetSnapshot() const override {
+        return Snapshot;
     }
 
-    void SetHasSnapshot(bool hasSnapshot) override {
-        ValidSnapshot = hasSnapshot;
+    void SetSnapshot(ui64 step, ui64 txId) override {
+        Snapshot.emplace();
+        Snapshot->SetStep(step);
+        Snapshot->SetTxId(txId);
     }
 
     bool BrokenLocks() const override {
-        return LocksIssue.has_value() && !(HasSnapshot() && IsReadOnly());
+        return LocksIssue.has_value() && !(GetSnapshot() && IsReadOnly());
     }
 
     const std::optional<NYql::TIssue>& GetLockIssue() const override {
@@ -405,8 +407,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 bool HasSnapshot() const = 0;
-    virtual void SetHasSnapshot(bool hasSnapshot) = 0;
+    virtual const std::optional<NKikimrDataEvents::TMvccSnapshot>& GetSnapshot() const = 0;
+    virtual void SetSnapshot(ui64 step, ui64 txId) = 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,10 +282,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
             }
         }
 
-        if (TxManager) {
-            TxManager->SetHasSnapshot(GetSnapshot().IsValid());
-        }
-
         if (!BufferActorId || (ReadOnlyTx && Request.LocksOp != ELocksOp::Rollback)) {
             Become(&TKqpDataExecuter::FinalizeState);
             MakeResponseAndPassAway();
@@ -2241,6 +2237,8 @@ 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();
@@ -2276,6 +2274,9 @@ 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/executer_actor/kqp_executer_impl.h

@@ -952,6 +952,10 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
             if (!settings.GetInconsistentTx() && !settings.GetIsOlap()) {
                 ActorIdToProto(BufferActorId, settings.MutableBufferActorId());
             }
+            if (GetSnapshot().IsValid() && settings.GetIsOlap()) {
+                settings.MutableMvccSnapshot()->SetStep(GetSnapshot().Step);
+                settings.MutableMvccSnapshot()->SetTxId(GetSnapshot().TxId);
+            }
             output.SinkSettings.ConstructInPlace();
             output.SinkSettings->PackFrom(settings);
         } else {

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -170,12 +170,14 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
         const bool inconsistentTx,
         const NMiniKQL::TTypeEnvironment& typeEnv,
         std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> alloc,
+        const NKikimrDataEvents::TMvccSnapshot& mvccSnapshot,
         const IKqpTransactionManagerPtr& txManager,
         const TActorId sessionActorId,
         TIntrusivePtr<TKqpCounters> counters,
         NWilson::TTraceId traceId)
         : TypeEnv(typeEnv)
         , Alloc(alloc)
+        , MvccSnapshot(mvccSnapshot)
         , TableId(tableId)
         , TablePath(tablePath)
         , LockTxId(lockTxId)
@@ -813,6 +815,7 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
             FillEvWritePrepare(evWrite.get(), shardId, *TxId, TxManager);
         } else if (!InconsistentTx) {
             evWrite->SetLockId(LockTxId, LockNodeId);
+            *evWrite->Record.MutableMvccSnapshot() = MvccSnapshot;
         }
 
         const auto serializationResult = ShardedWriteController->SerializeMessageToPayload(shardId, *evWrite);
@@ -954,6 +957,8 @@ class TKqpTableWriteActor : public TActorBootstrapped<TKqpTableWriteActor> {
     const NMiniKQL::TTypeEnvironment& TypeEnv;
     std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
 
+    NKikimrDataEvents::TMvccSnapshot MvccSnapshot;
+
     const TTableId TableId;
     const TString TablePath;
 
@@ -1019,6 +1024,7 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
             Settings.GetInconsistentTx(),
             TypeEnv,
             Alloc,
+            Settings.GetMvccSnapshot(),
             nullptr,
             TActorId{},
             Counters,
@@ -1206,6 +1212,7 @@ struct TTransactionSettings {
     ui64 LockTxId = 0;
     ui64 LockNodeId = 0;
     bool InconsistentTx = false;
+    NKikimrDataEvents::TMvccSnapshot MvccSnapshot;
 };
 
 struct TWriteSettings {
@@ -1316,10 +1323,14 @@ 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];
@@ -1333,6 +1344,7 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
                     InconsistentTx,
                     TypeEnv,
                     Alloc,
+                    MvccSnapshot,
                     TxManager,
                     SessionActorId,
                     Counters,
@@ -2011,6 +2023,7 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
 
     std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
     NMiniKQL::TTypeEnvironment TypeEnv;
+    NKikimrDataEvents::TMvccSnapshot MvccSnapshot;
 
     struct TWriteInfo {
         TKqpTableWriteActor* WriteTableActor = nullptr;
@@ -2137,6 +2150,7 @@ class TKqpForwardWriteActor : public TActorBootstrapped<TKqpForwardWriteActor>,
                     .LockTxId = Settings.GetLockTxId(),
                     .LockNodeId = Settings.GetLockNodeId(),
                     .InconsistentTx = Settings.GetInconsistentTx(),
+                    .MvccSnapshot = Settings.GetMvccSnapshot(),
                 },
                 .Priority = Settings.GetPriority(),
             };

ydb/core/protos/kqp.proto

@@ -719,6 +719,7 @@ message TKqpTableSinkSettings {
     optional NActorsProto.TActorId BufferActorId = 10;
     optional int64 Priority = 11;
     optional bool IsOlap = 12;
+    optional NKikimrDataEvents.TMvccSnapshot MvccSnapshot = 13;
 }
 
 message TKqpStreamLookupSettings {