25-3: Fix successful reads not always waiting for read conflicts to persist (#28060)

Author: maximyurchuk

ydb/core/tx/datashard/datashard__read_iterator.cpp

@@ -2953,7 +2953,8 @@ class TDataShard::TTxReadContinue : public NTabletFlatExecutor::TTransactionBase
 
             ApplyLocks(ctx);
 
-            if (Reader->NeedVolatileWaitForCommit() ||
+            if (txc.DB.HasChanges() ||
+                Reader->NeedVolatileWaitForCommit() ||
                 Self->Pipeline.HasCommittingOpsBelow(state.ReadVersion) ||
                 Self->GetVolatileTxManager().HasUnstableVolatileTxsAtSnapshot(state.ReadVersion))
             {

ydb/core/tx/datashard/datashard_ut_common_kqp.h

@@ -211,9 +211,13 @@ namespace NKqpHelpers {
         return KqpSimpleBeginWait(runtime, txId, KqpSimpleBeginSend(runtime, sessionId, query, database));
     }
 
-    inline TString KqpSimpleContinue(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query, const TString& database = {}) {
+    inline auto KqpSimpleContinueSend(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query, const TString& database = {}) {
         Y_ENSURE(!txId.empty(), "continue on empty transaction");
-        auto response = AwaitResponse(runtime, SendRequest(runtime, MakeSimpleRequestRPC(query, sessionId, txId, false /* commitTx */), database));
+        return SendRequest(runtime, MakeSimpleRequestRPC(query, sessionId, txId, false /* commitTx */), database);
+    }
+
+    inline TString KqpSimpleContinueWait(TTestActorRuntime& runtime, const TString& txId, NThreading::TFuture<Ydb::Table::ExecuteDataQueryResponse> future) {
+        auto response = AwaitResponse(runtime, std::move(future));
         if (response.operation().status() != Ydb::StatusIds::SUCCESS) {
             return TStringBuilder() << "ERROR: " << response.operation().status();
         }
@@ -223,6 +227,10 @@ namespace NKqpHelpers {
         return FormatResult(result);
     }
 
+    inline TString KqpSimpleContinue(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query, const TString& database = {}) {
+        return KqpSimpleContinueWait(runtime, txId, KqpSimpleContinueSend(runtime, sessionId, txId, query, database));
+    }
+
     inline auto KqpSimpleSendCommit(TTestActorRuntime& runtime, const TString& sessionId, const TString& txId, const TString& query, const TString& database = {}) {
         Y_ENSURE(!txId.empty(), "commit on empty transaction");
         return SendRequest(runtime, MakeSimpleRequestRPC(query, sessionId, txId, true /* commitTx */), database);

ydb/core/tx/datashard/datashard_ut_read_iterator.cpp

@@ -5162,6 +5162,126 @@ Y_UNIT_TEST_SUITE(DataShardReadIteratorConsistency) {
         );
     }
 
+    Y_UNIT_TEST(WriteLockUncommittedConflictFailure) {
+        TPortManager pm;
+        TServerSettings serverSettings(pm.GetPort(2134));
+        serverSettings.SetDomainName("Root")
+            .SetNodeCount(1)
+            .SetUseRealThreads(false);
+
+        // The bug requires restoring lock from persistent storage
+        serverSettings.FeatureFlags.SetEnableDataShardInMemoryStateMigration(false);
+
+        TServer::TPtr server = new TServer(serverSettings);
+
+        auto& runtime = *server->GetRuntime();
+        auto sender = runtime.AllocateEdgeActor();
+
+        runtime.SetLogPriority(NKikimrServices::TX_DATASHARD, NLog::PRI_TRACE);
+
+        InitRoot(server, sender);
+
+        TDisableDataShardLogBatching disableDataShardLogBatching;
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            KqpSchemeExec(runtime, R"(
+                CREATE TABLE `/Root/table` (key int, value int, PRIMARY KEY (key));
+            )"),
+            "SUCCESS"
+        );
+
+        const auto shards = GetTableShards(server, sender, "/Root/table");
+        UNIT_ASSERT_VALUES_EQUAL(shards.size(), 1u);
+
+        // Populate tables with initial values
+        ExecSQL(server, sender, R"(
+            UPSERT INTO `/Root/table` (key, value) VALUES
+                (1, 1001),
+                (2, 1002);
+        )");
+
+        // Block reads so read locks are not established yet
+        TBlockEvents<TEvDataShard::TEvRead> blockedReads(runtime);
+
+        // Begin the 1st transaction upserting to and reading from table
+        TString session1, tx1;
+        auto future1 = KqpSimpleBeginSend(runtime, session1, R"(
+                UPSERT INTO `/Root/table` (key, value) VALUES (3, 3003);
+                SELECT key, value FROM `/Root/table` ORDER BY key;
+            )");
+
+        // Begin the 2nd transaction upserting to and reading from table
+        TString session2, tx2;
+        auto future2 = KqpSimpleBeginSend(runtime, session2, R"(
+                UPSERT INTO `/Root/table` (key, value) VALUES (4, 4004);
+                SELECT key, value FROM `/Root/table` ORDER BY key;
+            )");
+
+        // Wait until both transactions are stuck at their read phases
+        runtime.WaitFor("blocked TEvRead x2", [&]{ return blockedReads.size() >= 2; });
+
+        // Force read iterators to read rows one-by-one
+        auto changeMaxRowsObserver = runtime.AddObserver<TEvDataShard::TEvRead>(
+            [&](auto& ev) {
+                auto* msg = ev->Get();
+                msg->Record.SetMaxRowsInResult(1);
+            });
+
+        // Block TEvReadContinue so it stops after the first result
+        TBlockEvents<TEvDataShard::TEvReadContinue> blockedReadContinue(runtime);
+
+        // Unblock reads and wait for corresponding TEvReadContinue
+        blockedReads.Stop().Unblock();
+
+        runtime.WaitFor("blocked TEvReadContinue x2", [&]{ return blockedReadContinue.size() >= 2; });
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        // Block further commits
+        TBlockEvents<TEvBlobStorage::TEvPut> blockedCommits(runtime, [&](auto& ev) {
+            auto* msg = ev->Get();
+            if (msg->Id.Channel() == 0 && msg->Id.TabletID() == shards.at(0)) {
+                Cerr << "... blocking commit " << msg->Id << Endl;
+                return true;
+            }
+            return false;
+        });
+
+        // Unblock TEvReadContinue and let kqp potentially receive results
+        blockedReadContinue.Stop().Unblock();
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        // Stop blocking commits and make them fail (will cause shards to restart)
+        blockedCommits.Stop();
+        for (auto& ev : blockedCommits) {
+            auto proxy = ev->Recipient;
+            ui32 groupId = GroupIDFromBlobStorageProxyID(proxy);
+            auto response = ev->Get()->MakeErrorResponse(NKikimrProto::ERROR, "Something went wrong", TGroupId::FromValue(groupId));
+            runtime.Send(new IEventHandle(ev->Sender, proxy, response.release()), 0, true);
+        }
+        runtime.SimulateSleep(TDuration::Seconds(1));
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            KqpSimpleBeginWait(runtime, tx1, std::move(future1)),
+            "{ items { int32_value: 1 } items { int32_value: 1001 } }, "
+            "{ items { int32_value: 2 } items { int32_value: 1002 } }, "
+            "{ items { int32_value: 3 } items { int32_value: 3003 } }");
+        UNIT_ASSERT_VALUES_EQUAL(
+            KqpSimpleBeginWait(runtime, tx2, std::move(future2)),
+            "{ items { int32_value: 1 } items { int32_value: 1001 } }, "
+            "{ items { int32_value: 2 } items { int32_value: 1002 } }, "
+            "{ items { int32_value: 4 } items { int32_value: 4004 } }");
+
+        // Commit the 1st transaction, it should succeed
+        UNIT_ASSERT_VALUES_EQUAL(
+            KqpSimpleCommit(runtime, session1, tx1, R"(SELECT 1)"),
+            "{ items { int32_value: 1 } }");
+
+        // Try to commit the 2nd transaction, it must fail
+        UNIT_ASSERT_VALUES_EQUAL(
+            KqpSimpleCommit(runtime, session2, tx2, R"(SELECT 1)"),
+            "ERROR: ABORTED");
+    }
+
 }
 
 Y_UNIT_TEST_SUITE(DataShardReadIteratorLatency) {