Move the breaking of the conflicting tx locks from CommitOnComplete to CommitOnExecute (#26206)

Co-authored-by: Kirill Vasilenko <[email protected]>

Author: kirillvasilenko

ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp

@@ -118,6 +118,7 @@ void TTxBlobsWritingFinished::DoComplete(const TActorContext& ctx) {
         pathIds.emplace(op->GetPathId().InternalPathId);
         if (op->GetBehaviour() == EOperationBehaviour::WriteWithLock || op->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
             if (op->GetBehaviour() != EOperationBehaviour::NoTxWrite || Self->GetOperationsManager().HasReadLocks(writeMeta.GetPathId().InternalPathId)) {
+                // detect active reads which the given tx has conflicts with
                 auto evWrite = std::make_shared<NOlap::NTxInteractions::TEvWriteWriter>(
                     writeMeta.GetPathId().InternalPathId, writeResult.GetPKBatchVerified(), Self->GetIndexOptional()->GetVersionedIndex().GetPrimaryKey());
                 Self->GetOperationsManager().AddEventForLock(*Self, op->GetLockId(), evWrite);
@@ -127,9 +128,11 @@ void TTxBlobsWritingFinished::DoComplete(const TActorContext& ctx) {
     auto& index = Self->MutableIndexAs<NOlap::TColumnEngineForLogs>();
     auto& granule = index.MutableGranuleVerified(Pack.GetPathId());
     for (auto&& portion : Pack.GetPortions()) {
+        // make the portions visible as uncommitted for reads
         granule.InsertPortionOnComplete(portion.GetPortionInfoPtr(), index);
     }
     if (PackBehaviour == EOperationBehaviour::NoTxWrite) {
+        // make the portions visible as committed for reads
         for (auto&& i : InsertWriteIds) {
             granule.CommitPortionOnComplete(i, index);
         }
@@ -139,7 +142,10 @@ void TTxBlobsWritingFinished::DoComplete(const TActorContext& ctx) {
         auto op = Self->GetOperationsManager().GetOperationVerified((TOperationWriteId)writeMeta.GetWriteId());
         if (op->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
             AFL_VERIFY(CommitSnapshot);
+            // No tx writes (bulk upsert) must break decent/proper txs.
+            // Decent/proper txs asked for serializable, so we have to give them serializable. 
             Self->OperationsManager->AddTemporaryTxLink(op->GetLockId());
+            Self->OperationsManager->BreakConflictingTxs(op->GetLockId());
             Self->OperationsManager->CommitTransactionOnComplete(*Self, op->GetLockId(), *CommitSnapshot);
             Self->Counters.GetTabletCounters()->IncCounter(COUNTER_IMMEDIATE_TX_COMPLETED);
         } else {

ydb/core/tx/columnshard/operations/manager.cpp

@@ -71,6 +71,25 @@ bool TOperationsManager::Load(NTabletFlatExecutor::TTransactionContext& txc) {
     return true;
 }
 
+void TOperationsManager::BreakConflictingTxs(const TLockFeatures& lock) {
+    for (auto&& i : lock.GetBrokeOnCommit()) {
+        if (auto lockNotify = GetLockOptional(i)) {
+            AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("broken_lock_id", i);
+            lockNotify->SetBroken();
+        }
+    }
+    for (auto&& i : lock.GetNotifyOnCommit()) {
+        if (auto lockNotify = GetLockOptional(i)) {
+            lockNotify->AddNotifyCommit(lock.GetLockId());
+        }
+    }
+}
+
+void TOperationsManager::BreakConflictingTxs(const ui64 txId) {
+    auto& lock = GetLockFeaturesForTxVerified(txId);
+    BreakConflictingTxs(lock);
+}
+
 void TOperationsManager::CommitTransactionOnExecute(
     TColumnShard& owner, const ui64 txId, NTabletFlatExecutor::TTransactionContext& txc, const NOlap::TSnapshot& snapshot) {
     auto& lock = GetLockFeaturesForTxVerified(txId);
@@ -81,6 +100,9 @@ void TOperationsManager::CommitTransactionOnExecute(
         opPtr->CommitOnExecute(owner, txc, snapshot);
         commited.emplace_back(opPtr);
     }
+
+    BreakConflictingTxs(lock);
+
     OnTransactionFinishOnExecute(commited, lock, txId, txc);
 }
 
@@ -89,18 +111,6 @@ void TOperationsManager::CommitTransactionOnComplete(
     auto& lock = GetLockFeaturesForTxVerified(txId);
     TLogContextGuard gLogging(
         NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_TX)("commit_tx_id", txId)("commit_lock_id", lock.GetLockId()));
-    for (auto&& i : lock.GetBrokeOnCommit()) {
-        if (auto lockNotify = GetLockOptional(i)) {
-            AFL_WARN(NKikimrServices::TX_COLUMNSHARD_TX)("broken_lock_id", i);
-            lockNotify->SetBroken();
-        }
-    }
-
-    for (auto&& i : lock.GetNotifyOnCommit()) {
-        if (auto lockNotify = GetLockOptional(i)) {
-            lockNotify->AddNotifyCommit(lock.GetLockId());
-        }
-    }
 
     TVector<TWriteOperation::TPtr> commited;
     for (auto&& opPtr : lock.GetWriteOperations()) {
@@ -317,10 +327,12 @@ void TOperationsManager::AddEventForLock(
     NOlap::NTxInteractions::TTxConflicts txConflicts;
     auto& txLock = GetLockVerified(lockId);
     writer->CheckInteraction(lockId, InteractionsContext, txConflicts, txNotifications);
-    for (auto&& i : txConflicts) {
-        if (GetLockOptional(i.first)) {
-            GetLockVerified(i.first).AddBrokeOnCommit(i.second);
-        } else if (txLock.IsCommitted(i.first)) {
+    for (auto& [commitLockId, breakLockIds] : txConflicts) {
+        if (GetLockOptional(commitLockId)) {
+            GetLockVerified(commitLockId).AddBrokeOnCommit(breakLockIds);
+        // if txId not found, it means the conflicting tx is already committed or aborted
+        } else if (txLock.IsCommitted(commitLockId)) {
+            // if the conflicting tx is already committed, we cannot commit the given tx, so break its lock
             txLock.SetBroken();
         }
     }

ydb/core/tx/columnshard/operations/manager.h

@@ -236,6 +236,9 @@ class TOperationsManager {
     void AbortLockOnExecute(TColumnShard& owner, const ui64 lockId, NTabletFlatExecutor::TTransactionContext& txc);
     void AbortLockOnComplete(TColumnShard& owner, const ui64 lockId);
 
+    void BreakConflictingTxs(const TLockFeatures& lock);
+    void BreakConflictingTxs(const ui64 txId);
+
     std::optional<ui64> GetLockForTx(const ui64 txId) const;
     std::optional<ui64> GetLockForTxOptional(const ui64 txId) const {
         return GetLockForTx(txId);