correct generations usage for ignore deprecated sessions (#4563)

Author: ivanmorozov333

ydb/core/kqp/compute_actor/kqp_scan_compute_manager.h

@@ -24,6 +24,7 @@ class TShardScannerInfo {
 private:
     std::optional<TActorId> ActorId;
     const ui64 TabletId;
+    const ui64 Generation;
     i64 DataChunksInFlightCount = 0;
     bool TracingStarted = false;
     const ui64 FreeSpace = (ui64)8 << 20;
@@ -46,18 +47,19 @@ class TShardScannerInfo {
             TracingStarted = true;
         }
         if (NActors::TlsActivationContext) {
-            NActors::TActivationContext::AsActorContext().Send(*ActorId, new TEvKqpCompute::TEvScanDataAck(FreeSpace, TabletId, 1), flags, TabletId);
+            NActors::TActivationContext::AsActorContext().Send(*ActorId, new TEvKqpCompute::TEvScanDataAck(FreeSpace, Generation, 1), flags, TabletId);
         }
     }
 public:
     TShardScannerInfo(TShardState& state, const IExternalObjectsProvider& externalObjectsProvider)
         : TabletId(state.TabletId)
+        , Generation(++state.Generation)
     {
         const bool subscribed = std::exchange(state.SubscribedOnTablet, true);
 
         const auto& keyColumnTypes = externalObjectsProvider.GetKeyColumnTypes();
         auto ranges = state.GetScanRanges(keyColumnTypes);
-        auto ev = externalObjectsProvider.BuildEvKqpScan(0, TabletId, ranges);
+        auto ev = externalObjectsProvider.BuildEvKqpScan(0, Generation, ranges);
 
         AFL_DEBUG(NKikimrServices::KQP_COMPUTE)("event", "start_scanner")("info", state.ToString(keyColumnTypes))
             ("range", DebugPrintRanges(keyColumnTypes, ranges, *AppData()->TypeRegistry))("subscribed", subscribed);
@@ -278,9 +280,9 @@ class TInFlightShards: public NComputeActor::TScanShardsStatistics {
         }
     }
 
-    void RegisterScannerActor(const ui64 tabletId, const TActorId& scanActorId) {
+    void RegisterScannerActor(const ui64 tabletId, const ui64 generation, const TActorId& scanActorId) {
         auto state = GetShardState(tabletId);
-        if (!state) {
+        if (!state || generation != state->Generation) {
             return;
         }
         AFL_ENSURE(state->State == NComputeActor::EShardState::Starting)("state", state->State);

ydb/core/kqp/compute_actor/kqp_scan_fetcher_actor.cpp

@@ -106,15 +106,15 @@ void TKqpScanFetcherActor::HandleExecute(TEvKqpCompute::TEvScanInitActor::TPtr&
     }
     auto& msg = ev->Get()->Record;
     auto scanActorId = ActorIdFromProto(msg.GetScanActorId());
-    InFlightShards.RegisterScannerActor(msg.GetTabletId(), scanActorId);
+    InFlightShards.RegisterScannerActor(msg.GetTabletId(), msg.GetGeneration(), scanActorId);
 }
 
 void TKqpScanFetcherActor::HandleExecute(TEvKqpCompute::TEvScanData::TPtr& ev) {
     if (!InFlightShards.IsActive()) {
         return;
     }
     auto state = InFlightShards.GetShardStateByActorId(ev->Sender);
-    if (!state) {
+    if (!state || state->Generation != ev->Get()->Generation) {
         return;
     }
     AFL_ENSURE(state->State == EShardState::Running)("state", state->State)("actor_id", state->ActorId)("ev_sender", ev->Sender);
@@ -152,6 +152,9 @@ void TKqpScanFetcherActor::HandleExecute(TEvKqpCompute::TEvScanError::TPtr& ev)
             return;
         }
     }
+    if (state->Generation != ev->Get()->Record.GetGeneration()) {
+        return;
+    }
 
     if (state->State == EShardState::Starting) {
         ++TotalRetries;