Merge 3ae5370 into d543026

ydb-platform · Dec 26, 2024 · 8236fcd · 8236fcd
2 parents d543026 + 3ae5370
commit 8236fcd
Showing 1 changed file with 11 additions and 6 deletions.
diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -795,6 +795,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
 
     void BeginTx(const Ydb::Table::TransactionSettings& settings) {
         QueryState->TxId.SetValue(UlidGen.Next());
+        TerminateBufferActor();
         QueryState->TxCtx = MakeIntrusive<TKqpTransactionContext>(false, AppData()->FunctionRegistry,
             AppData()->TimeProvider, AppData()->RandomProvider);
 
@@ -881,6 +882,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
                     break;
             }
         } else {
+            TerminateBufferActor();
             QueryState->TxCtx = MakeIntrusive<TKqpTransactionContext>(false, AppData()->FunctionRegistry,
                 AppData()->TimeProvider, AppData()->RandomProvider);
             QueryState->QueryData = std::make_shared<TQueryData>(QueryState->TxCtx->TxAlloc);
@@ -2207,12 +2209,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
             QueryState->TxCtx->ClearDeferredEffects();
             QueryState->TxCtx->Locks.Clear();
             QueryState->TxCtx->TxManager.reset();
-
-            if (QueryState->TxCtx->BufferActorId) {
-                Send(QueryState->TxCtx->BufferActorId, new TEvKqpBuffer::TEvTerminate{});
-                QueryState->TxCtx->BufferActorId = {};
-            }
-
+            TerminateBufferActor();
             QueryState->TxCtx->Finish();
         }
     }
@@ -2230,6 +2227,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
                 Transactions.AddToBeAborted(txCtx);
                 Transactions.ReleaseTransaction(QueryState->TxId.GetValue());
             }
+            TerminateBufferActor();
             DiscardPersistentSnapshot(txCtx->SnapshotHandle);
         }
 
@@ -2702,6 +2700,13 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         QueryState->TxCtx->TopicOperations.SetWriteId(std::move(handle));
     }
 
+    void TerminateBufferActor() {
+        if (QueryState && QueryState->TxCtx && QueryState->TxCtx->BufferActorId) {
+            Send(QueryState->TxCtx->BufferActorId, new TEvKqpBuffer::TEvTerminate{});
+            QueryState->TxCtx->BufferActorId = {};
+        }
+    }
+
 private:
     TActorId Owner;
     TKqpQueryCachePtr QueryCache;