Flash: add MPP sql digest and connection metadata tracing

dbms/src/Flash/Coprocessor/DAGContext.cpp

@@ -104,6 +104,8 @@ DAGContext::DAGContext(tipb::DAGRequest & dag_request_, const mpp::TaskMeta & me
     , resource_group_name(meta_.resource_group_name())
     , connection_id(meta_.connection_id())
     , connection_alias(meta_.connection_alias())
+    , sql_digest(meta_.sql_digest())
+    , plan_digest(meta_.plan_digest())
 {
     if (dag_request->has_div_precision_increment())
         div_precision_increment = dag_request->div_precision_increment();
@@ -483,7 +485,7 @@ UInt64 DAGContext::getReadBytes() const
     UInt64 read_bytes = 0;
     for (const auto & [id, sc] : scan_context_map)
     {
-        (void)id; // Disable unused variable warnning.
+        (void)id; // Disable unused variable warning.
         read_bytes += sc->userReadBytes();
     }
     return read_bytes;

dbms/src/Flash/Coprocessor/DAGContext.h

@@ -361,6 +361,8 @@ class DAGContext
 
     UInt64 getConnectionID() const { return connection_id; }
     const String & getConnectionAlias() const { return connection_alias; }
+    const String & getSQLDigest() const { return sql_digest; }
+    const String & getPlanDigest() const { return plan_digest; }
 
     MPPReceiverSetPtr getMPPReceiverSet() const { return mpp_receiver_set; }
 
@@ -544,6 +546,8 @@ class DAGContext
     UInt64 connection_id;
     // It's the session alias between mysql client and tidb
     String connection_alias;
+    String sql_digest;
+    String plan_digest;
 
     String query_id_and_cte_id_for_sink;
     std::unordered_map<size_t, String> query_id_and_cte_id_for_sources;

dbms/src/Flash/FlashService.cpp

@@ -495,11 +495,14 @@ grpc::Status FlashService::DispatchMPPTask(
     const auto & resource_group = task_meta.resource_group_name();
     LOG_INFO(
         log,
-        "Handling mpp dispatch request, task: {}, resource_group: {}, conn_id: {}, conn_alias: {}",
+        "Handling mpp dispatch request, task: {}, resource_group: {}, conn_id: {}, conn_alias: {}, "
+        "sql_digest: {}, plan_digest: {}",
         MPPTaskId(task_meta).toString(),
         resource_group,
         task_meta.connection_id(),
-        task_meta.connection_alias());
+        task_meta.connection_alias(),
+        task_meta.sql_digest(),
+        task_meta.plan_digest());
     auto check_result = checkGrpcContext(grpc_context);
     if (!check_result.ok())
         return check_result;

dbms/src/Flash/Mpp/MPPTaskStatistics.cpp

@@ -77,6 +77,10 @@ void MPPTaskStatistics::initializeExecutorDAG(DAGContext * dag_context_)
 
     is_root = dag_context->isRootMPPTask();
     sender_executor_id = root_executor.executor_id();
+    connection_id = dag_context->getConnectionID();
+    connection_alias = dag_context->getConnectionAlias();
+    sql_digest = dag_context->getSQLDigest();
+    plan_digest = dag_context->getPlanDigest();
     executor_statistics_collector.initialize(dag_context);
 }
 
@@ -110,6 +114,7 @@ void MPPTaskStatistics::logTracingJson()
         /// don't use info log for initializing status since it does not contains too many information
         status == INITIALIZING ? Poco::Message::PRIO_DEBUG : Poco::Message::PRIO_INFORMATION,
         R"({{"query_tso":{},"task_id":{},"is_root":{},"sender_executor_id":"{}","executors":{},"host":"{}")"
+        R"(,"connection_id":{},"connection_alias":"{}","sql_digest":"{}","plan_digest":"{}")"
         R"(,"task_init_timestamp":{},"task_start_timestamp":{},"task_end_timestamp":{})"
         R"(,"compile_start_timestamp":{},"compile_end_timestamp":{})"
         R"(,"read_wait_index_start_timestamp":{},"read_wait_index_end_timestamp":{})"
@@ -121,6 +126,10 @@ void MPPTaskStatistics::logTracingJson()
         sender_executor_id,
         executor_statistics_collector.profilesToJson(),
         host,
+        connection_id,
+        connection_alias,
+        sql_digest,
+        plan_digest,
         toNanoseconds(task_init_timestamp),
         toNanoseconds(task_start_timestamp),
         toNanoseconds(task_end_timestamp),

dbms/src/Flash/Mpp/MPPTaskStatistics.h

@@ -89,6 +89,10 @@ class MPPTaskStatistics
     // executor dag
     bool is_root = false;
     String sender_executor_id;
+    UInt64 connection_id = 0;
+    String connection_alias;
+    String sql_digest;
+    String plan_digest;
 
     // resource
     RUConsumption ru_info{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0};