Fix INTERNAL ERROR instead of ABORTED for Snapshot Isolation (#46143)

2 files changed, 68 insertions, 0 deletions

diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
index 915518778a4..c4fa8627146 100644
--- a/ydb/core/kqp/session_actor/kqp_session_actor.cpp
+++ b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -1710,6 +1710,16 @@ public:
             return false;
         }
 
+        if (txCtx.TxManager->BrokenLocks()) {
+            EmitVictimTliLog(
+                txCtx.TxManager->GetVictimQuerySpanId(),
+                std::nullopt,
+                txCtx.TxManager->GetBrokenLocksCount());
+            ReplyQueryError(Ydb::StatusIds::ABORTED, "transaction locks invalidated",
+                MessageFromIssues(std::vector<TIssue>{*txCtx.TxManager->GetLockIssue()}));
+            return false;
+        }
+
         return true;
     }
 
diff --git a/ydb/core/kqp/ut/tx/kqp_snapshot_isolation_ut.cpp b/ydb/core/kqp/ut/tx/kqp_snapshot_isolation_ut.cpp
index e13ccac8257..35abd67a480 100644
--- a/ydb/core/kqp/ut/tx/kqp_snapshot_isolation_ut.cpp
+++ b/ydb/core/kqp/ut/tx/kqp_snapshot_isolation_ut.cpp
@@ -989,6 +989,64 @@ Y_UNIT_TEST_SUITE(KqpSnapshotIsolation) {
         tester.SetIsUpdate(IsUpdate);
         tester.Execute();
     }
+
+    class TBrokenLocksReadAfterUpsert : public TTableDataModificationTester {
+    protected:
+        void DoExecute() override {
+            auto client = Kikimr->GetQueryClient();
+            auto session1 = client.GetSession().GetValueSync().GetSession();
+            auto session2 = client.GetSession().GetValueSync().GetSession();
+
+            auto result = session1.ExecuteQuery(Q_(R"(
+                SELECT * FROM `/Root/Test` WHERE Name == "Paul";
+            )"), TTxControl::BeginTx(TTxSettings::SnapshotRW())).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+            auto tx1 = result.GetTransaction();
+            UNIT_ASSERT(tx1);
+            UNIT_ASSERT(tx1->IsActive());
+
+            result = session1.ExecuteQuery(Q_(R"(
+                PRAGMA kikimr.KqpForceImmediateEffectsExecution="true";
+                UPSERT INTO `/Root/Test` (Group, Name, Comment)
+                VALUES (1U, "Paul", "Changed");
+            )"), TTxControl::Tx(*tx1)).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+            result = session2.ExecuteQuery(Q_(R"(
+                UPSERT INTO `/Root/Test` (Group, Name, Comment)
+                VALUES (1U, "Paul", "Changed Other");
+            )"), TTxControl::BeginTx(TTxSettings::SnapshotRW()).CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+            result = session1.ExecuteQuery(Q_(R"(
+                SELECT * FROM `/Root/Test` WHERE Name == "Paul";
+            )"), TTxControl::Tx(*tx1)).ExtractValueSync();
+            if (GetIsOlap()) {
+                UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+            } else  {
+                UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::ABORTED, result.GetIssues().ToString());
+            }
+
+            if (GetIsOlap()) {
+                result = session1.ExecuteQuery(Q_(R"(
+                    SELECT * FROM `/Root/Test` WHERE Name == "Paul";
+                )"), TTxControl::Tx(*tx1)).ExtractValueSync();
+                UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::ABORTED, result.GetIssues().ToString());
+            }
+        }
+    };
+
+    Y_UNIT_TEST(TBrokenLocksReadAfterUpsertOltp) {
+        TBrokenLocksReadAfterUpsert tester;
+        tester.SetIsOlap(false);
+        tester.Execute();
+    }
+
+    Y_UNIT_TEST(TBrokenLocksReadAfterUpsertOlap) {
+        TBrokenLocksReadAfterUpsert tester;
+        tester.SetIsOlap(true);
+        tester.Execute();
+    }
 }
 
 } // namespace NKqp