Don't recreate snapshots in transaction (#11553)

Author: nikvas0

ydb/core/kqp/common/kqp_tx.cpp

@@ -171,7 +171,6 @@ bool NeedSnapshot(const TKqpTransactionContext& txCtx, const NYql::TKikimrConfig
 
     size_t readPhases = 0;
     bool hasEffects = false;
-    bool hasSourceRead = false;
     bool hasStreamLookup = false;
     bool hasSinkWrite = false;
 
@@ -191,7 +190,6 @@ bool NeedSnapshot(const TKqpTransactionContext& txCtx, const NYql::TKikimrConfig
         }
 
         for (const auto &stage : tx.GetStages()) {
-            hasSourceRead |= !stage.GetSources().empty();
             hasSinkWrite |= !stage.GetSinks().empty();
 
             for (const auto &input : stage.GetInputs()) {
@@ -211,9 +209,7 @@ bool NeedSnapshot(const TKqpTransactionContext& txCtx, const NYql::TKikimrConfig
         return true;
     }
 
-    if ((hasSourceRead || hasStreamLookup) && hasSinkWrite) {
-        return true;
-    }
+    YQL_ENSURE(!hasSinkWrite || hasEffects);
 
     // We don't want snapshot when there are effects at the moment,
     // because it hurts performance when there are multiple single-shard

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -191,6 +191,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
 
     bool ForceAcquireSnapshot() const {
         const bool forceSnapshot = (
+            !GetSnapshot().IsValid() &&
             ReadOnlyTx &&
             !ImmediateTx &&
             !HasPersistentChannels &&

ydb/core/kqp/runtime/kqp_read_actor.cpp

@@ -598,8 +598,8 @@ class TKqpReadActor : public TActorBootstrapped<TKqpReadActor>, public NYql::NDq
                 PendingShards.PushBack(state.Release());
                 return;
             }
-        } else if (!Snapshot.IsValid() && !Settings->GetAllowInconsistentReads()) {
-            return RuntimeError("Inconsistent reads after shards split", NDqProto::StatusIds::UNAVAILABLE);
+        } else if (!Snapshot.IsValid() && !Settings->HasLockTxId() && !Settings->GetAllowInconsistentReads()) {
+            return RuntimeError("Inconsistent reads without locks", NDqProto::StatusIds::UNAVAILABLE);
         }
 
         const auto& tr = *AppData()->TypeRegistry;