Fix report status error

dbms/src/Flash/Mpp/MPPTaskStatistics.cpp

@@ -99,11 +99,13 @@ void MPPTaskStatistics::collectRuntimeStatistics()
 
 tipb::SelectResponse MPPTaskStatistics::genExecutionSummaryResponse()
 {
+    syncRUInfoToExecutorStatisticsCollector();
     return executor_statistics_collector.genExecutionSummaryResponse();
 }
 
 tipb::TiFlashExecutionInfo MPPTaskStatistics::genTiFlashExecutionInfo()
 {
+    syncRUInfoToExecutorStatisticsCollector();
     return executor_statistics_collector.genTiFlashExecutionInfo();
 }
 
@@ -153,10 +155,15 @@ void MPPTaskStatistics::setMemoryPeak(Int64 memory_peak_)
     memory_peak = memory_peak_;
 }
 
+void MPPTaskStatistics::syncRUInfoToExecutorStatisticsCollector()
+{
+    executor_statistics_collector.setLocalRUConsumption(ru_info);
+}
+
 void MPPTaskStatistics::setRUInfo(const RUConsumption & ru_info_)
 {
     ru_info = ru_info_;
-    executor_statistics_collector.setLocalRUConsumption(ru_info_);
+    syncRUInfoToExecutorStatisticsCollector();
 }
 
 void MPPTaskStatistics::setCompileTimestamp(const Timestamp & start_timestamp, const Timestamp & end_timestamp)

dbms/src/Flash/Mpp/MPPTaskStatistics.h

@@ -61,6 +61,8 @@ class MPPTaskStatistics
     tipb::TiFlashExecutionInfo genTiFlashExecutionInfo();
 
 private:
+    void syncRUInfoToExecutorStatisticsCollector();
+
     void recordInputBytes(DAGContext & dag_context);
 
     const LoggerPtr log;

dbms/src/Flash/Statistics/ExecutorStatisticsCollector.cpp

@@ -156,7 +156,14 @@ tipb::TiFlashExecutionInfo ExecutorStatisticsCollector::genTiFlashExecutionInfo(
 
 void ExecutorStatisticsCollector::setLocalRUConsumption(const RUConsumption & ru_info)
 {
-    local_ru = std::make_optional<resource_manager::Consumption>();
+    if (!local_ru)
+    {
+        local_ru.emplace();
+    }
+    else
+    {
+        local_ru->Clear();
+    }
     local_ru->set_r_r_u(ru_info.cpu_ru + ru_info.read_ru);
     local_ru->set_total_cpu_time_ms(toCPUTimeMillisecond(ru_info.cpu_time_ns));
     local_ru->set_read_bytes(ru_info.read_bytes);

dbms/src/Flash/tests/gtest_execution_summary.cpp

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <Flash/Mpp/MPPTaskStatistics.h>
+#include <Flash/Mpp/MPPTunnelSet.h>
 #include <Interpreters/Context.h>
 #include <TestUtils/ExecutorTestUtils.h>
 #include <TestUtils/mockExecutor.h>
+#include <kvproto/resource_manager.pb.h>
 
 namespace DB
 {
@@ -150,6 +153,48 @@ try
 }
 CATCH
 
+TEST_F(ExecutionSummaryTestRunner, genMPPTaskExecutionInfoWithoutSetRUInfo)
+try
+{
+    auto request = context.scan("test_db", "test_table").limit(1).exchangeSender(tipb::PassThrough).build(context);
+    request->set_collect_execution_summaries(true);
+
+    mpp::TaskMeta meta;
+    meta.set_gather_id(1);
+    meta.set_task_id(1);
+    meta.set_query_ts(1);
+    meta.set_local_query_id(1);
+    meta.set_server_id(1);
+    meta.set_start_ts(1);
+
+    DAGContext dag_context(*request, meta, false);
+    // ExchangeSenderStatistics reads tunnel_set while initializing the mock MPP DAG.
+    dag_context.tunnel_set = std::make_shared<MPPTunnelSet>("test-host");
+    MPPTaskStatistics task_statistics(MPPTaskId(meta), "test-host");
+    task_statistics.initializeExecutorDAG(&dag_context);
+
+    auto execution_info = task_statistics.genTiFlashExecutionInfo();
+    const auto root_executor_id = dag_context.dag_request.rootExecutorID();
+    const tipb::ExecutorExecutionSummary * root_summary = nullptr;
+    for (const auto & summary : execution_info.execution_summaries())
+    {
+        if (summary.executor_id() == root_executor_id)
+        {
+            root_summary = &summary;
+            break;
+        }
+    }
+
+    ASSERT_NE(root_summary, nullptr);
+    ASSERT_TRUE(root_summary->has_ru_consumption());
+    resource_manager::Consumption ru_consumption;
+    ASSERT_TRUE(ru_consumption.ParseFromString(root_summary->ru_consumption()));
+    ASSERT_DOUBLE_EQ(0.0, ru_consumption.r_r_u());
+    ASSERT_DOUBLE_EQ(0.0, ru_consumption.read_bytes());
+    ASSERT_DOUBLE_EQ(0.0, ru_consumption.total_cpu_time_ms());
+}
+CATCH
+
 TEST_F(ExecutionSummaryTestRunner, treeBased)
 try
 {