From 785712432094f9e0d3354a7ebd5bc2dc6d16f4c9 Mon Sep 17 00:00:00 2001
From: jinhelin <linjinhe33@gmail.com>
Date: Mon, 29 Aug 2022 21:48:23 +0800
Subject: [PATCH] Use read tso instead of threrad ID in 'DBGInvoke
 search_log_for_key' (#5715)

close pingcap/tiflash#5718
---
 dbms/src/Debug/dbgFuncMisc.cpp          | 26 ++++++++++++-------------
 dbms/src/Storages/StorageDeltaMerge.cpp |  5 +++--
 2 files changed, 16 insertions(+), 15 deletions(-)

diff --git a/dbms/src/Debug/dbgFuncMisc.cpp b/dbms/src/Debug/dbgFuncMisc.cpp
index e43ca5dd725..a1d592f88e9 100644
--- a/dbms/src/Debug/dbgFuncMisc.cpp
+++ b/dbms/src/Debug/dbgFuncMisc.cpp
@@ -23,13 +23,13 @@
 
 namespace DB
 {
-inline size_t getThreadIdForLog(const String & line)
+inline size_t getReadTSOForLog(const String & line)
 {
-    auto sub_line = line.substr(line.find("thread_id="));
+    auto sub_line = line.substr(line.find("read_tso="));
     std::regex rx(R"((0|[1-9][0-9]*))");
     std::smatch m;
     if (regex_search(sub_line, m, rx))
-        return std::stoi(m[1]);
+        return std::stoul(m[1]);
     else
         return 0;
 }
@@ -49,33 +49,33 @@ void dbgFuncSearchLogForKey(Context & context, const ASTs & args, DBGInvoker::Pr
     if (args.size() < 2)
         throw Exception("Args not matched, should be: key, thread_hint", ErrorCodes::BAD_ARGUMENTS);
 
-    String key = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    auto key = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
     // the candidate line must be printed by a thread which also print a line contains `thread_hint`
-    String thread_hint = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[1]).value);
+    auto tso_hint = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[1]).value);
     auto log_path = context.getConfigRef().getString("logger.log");
 
     std::ifstream file(log_path);
     // get the lines containing `thread_hint` and `key`
-    std::vector<String> thread_hint_line_candidates;
+    std::vector<String> tso_hint_line_candidates;
     std::vector<String> key_line_candidates;
     {
         String line;
         while (std::getline(file, line))
         {
-            if ((line.find(thread_hint) != String::npos) && (line.find("DBGInvoke") == String::npos))
-                thread_hint_line_candidates.emplace_back(line);
+            if ((line.find(tso_hint) != String::npos) && (line.find("DBGInvoke") == String::npos))
+                tso_hint_line_candidates.emplace_back(line);
             else if ((line.find(key) != String::npos) && (line.find("DBGInvoke") == String::npos))
                 key_line_candidates.emplace_back(line);
         }
     }
-    // get target thread id
-    if (thread_hint_line_candidates.empty() || key_line_candidates.empty())
+    // get target read tso
+    if (tso_hint_line_candidates.empty() || key_line_candidates.empty())
     {
         output("Invalid");
         return;
     }
-    size_t target_thread_id = getThreadIdForLog(thread_hint_line_candidates.back());
-    if (target_thread_id == 0)
+    size_t target_read_tso = getReadTSOForLog(tso_hint_line_candidates.back());
+    if (target_read_tso == 0)
     {
         output("Invalid");
         return;
@@ -83,7 +83,7 @@ void dbgFuncSearchLogForKey(Context & context, const ASTs & args, DBGInvoker::Pr
     String target_line;
     for (auto iter = key_line_candidates.rbegin(); iter != key_line_candidates.rend(); iter++)
     {
-        if (getThreadIdForLog(*iter) == target_thread_id)
+        if (getReadTSOForLog(*iter) == target_read_tso)
         {
             target_line = *iter;
             break;
diff --git a/dbms/src/Storages/StorageDeltaMerge.cpp b/dbms/src/Storages/StorageDeltaMerge.cpp
index d908c9a46ed..485c0057135 100644
--- a/dbms/src/Storages/StorageDeltaMerge.cpp
+++ b/dbms/src/Storages/StorageDeltaMerge.cpp
@@ -646,7 +646,8 @@ BlockInputStreams StorageDeltaMerge::read(
         throw Exception("TMTContext is not initialized", ErrorCodes::LOGICAL_ERROR);
 
     const auto & mvcc_query_info = *query_info.mvcc_query_info;
-    auto tracing_logger = Logger::get("StorageDeltaMerge", log->identifier(), query_info.req_id);
+    auto req_id = fmt::format("{} read_tso={}", query_info.req_id, mvcc_query_info.read_tso);
+    auto tracing_logger = Logger::get("StorageDeltaMerge", log->identifier(), req_id);
 
     LOG_FMT_DEBUG(tracing_logger, "Read with tso: {}", mvcc_query_info.read_tso);
 
@@ -756,7 +757,7 @@ BlockInputStreams StorageDeltaMerge::read(
         num_streams,
         /*max_version=*/mvcc_query_info.read_tso,
         rs_operator,
-        query_info.req_id,
+        req_id,
         query_info.keep_order,
         /* is_fast_mode */ tidb_table_info.tiflash_mode == TiDB::TiFlashMode::Fast, // read in normal mode or read in fast mode
         max_block_size,