Flash: add MPP plan digest

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

dbms/src/Flash/Coprocessor/DAGContext.cpp

@@ -105,6 +105,7 @@ DAGContext::DAGContext(tipb::DAGRequest & dag_request_, const mpp::TaskMeta & me
     , 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();

dbms/src/Flash/Coprocessor/DAGContext.h

@@ -362,6 +362,7 @@ 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; }
 
@@ -546,6 +547,7 @@ class DAGContext
     // 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,12 +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: {}, sql_digest: {}",
+        "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.sql_digest());
+        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

@@ -80,6 +80,7 @@ void MPPTaskStatistics::initializeExecutorDAG(DAGContext * dag_context_)
     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);
 }
 
@@ -113,7 +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":"{}")"
+        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":{})"
@@ -128,6 +129,7 @@ void MPPTaskStatistics::logTracingJson()
         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

@@ -92,6 +92,7 @@ class MPPTaskStatistics
     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};