executor: report join stats to TiDB

dbms/src/Flash/Coprocessor/DAGContext.h

@@ -74,6 +74,10 @@ struct JoinProfileInfo
     bool is_spilled = false;
     /// Empty only for a cross join, which does not have a hash table.
     std::optional<HashTableStats> hash_table_stats;
+    /// Total number of row pairs enumerated by this join's probes before other conditions are
+    /// applied. Unlike the hash table stats above, it is not tied to any hash table, so cross
+    /// joins report it too. Present once the probe finishes (0 means the join matched nothing).
+    std::optional<UInt64> joined_rows;
 
     void setHashTableStats(const HashTableStats & stats)
     {
@@ -90,12 +94,28 @@ struct JoinProfileInfo
             hash_table_stats->merge(stats);
     }
 
+    /// Unlike the size/memory fields which are finalized at build finish, joined rows are only
+    /// known when each probe stream finishes, so they are accumulated here after the build stage
+    /// has already published its own stats. Each Join (including every restore join) reports the
+    /// pairs enumerated by its own probe streams exactly once.
+    void addJoinedRows(UInt64 rows)
+    {
+        std::lock_guard lock(hash_table_stats_mutex);
+        joined_rows = joined_rows.value_or(0) + rows;
+    }
+
     std::optional<HashTableStats> getHashTableStats() const
     {
         std::lock_guard lock(hash_table_stats_mutex);
         return hash_table_stats;
     }
 
+    std::optional<UInt64> getJoinedRows() const
+    {
+        std::lock_guard lock(hash_table_stats_mutex);
+        return joined_rows;
+    }
+
 private:
     mutable std::mutex hash_table_stats_mutex;
 };

dbms/src/Flash/Coprocessor/ExecutionSummary.cpp

@@ -74,6 +74,8 @@ void ExecutionSummary::merge(const ExecutionSummary & other)
         else
             hash_table_stats->merge(*other.hash_table_stats);
     }
+    if (other.joined_rows)
+        joined_rows = joined_rows.value_or(0) + *other.joined_rows;
     ru_consumption = mergeRUConsumption(ru_consumption, other.ru_consumption);
     scan_context->merge(*other.scan_context);
     if (other.columnar_scan_context)
@@ -108,6 +110,8 @@ void ExecutionSummary::merge(const tipb::ExecutorExecutionSummary & other)
         else
             hash_table_stats->merge(other_hash_table_stats);
     }
+    if (other.has_tiflash_join_stats() && other.tiflash_join_stats().has_joined_rows())
+        joined_rows = joined_rows.value_or(0) + other.tiflash_join_stats().joined_rows();
     ru_consumption = mergeRUConsumption(ru_consumption, parseRUConsumption(other));
     if (other.has_tiflash_scan_context())
         scan_context->merge(other.tiflash_scan_context());
@@ -149,6 +153,8 @@ void ExecutionSummary::init(const tipb::ExecutorExecutionSummary & other)
     inter_zone_receive_bytes = other.tiflash_network_summary().inter_zone_receive_bytes();
     if (other.has_tiflash_hash_table_stats())
         hash_table_stats = toHashTableStats(other.tiflash_hash_table_stats());
+    if (other.has_tiflash_join_stats() && other.tiflash_join_stats().has_joined_rows())
+        joined_rows = other.tiflash_join_stats().joined_rows();
     ru_consumption = parseRUConsumption(other);
     if (other.has_tiflash_scan_context())
         scan_context->deserialize(other.tiflash_scan_context());

dbms/src/Flash/Coprocessor/ExecutionSummary.h

@@ -45,6 +45,9 @@ struct ExecutionSummary
     /// This summary belongs to one physical executor, identified by executor_id in the protobuf.
     /// Therefore each hash-table operator gets an independent value; merge only combines fragments of the same operator.
     std::optional<HashTableStats> hash_table_stats;
+    /// Row pairs enumerated by a join's probes before other conditions are applied; only join
+    /// executors report it, cross joins included. Absent when the probe never completes.
+    std::optional<UInt64> joined_rows;
     resource_manager::Consumption ru_consumption{};
 
     DM::ScanContextPtr scan_context;

dbms/src/Flash/Statistics/ExecutionSummaryHelper.cpp

@@ -42,6 +42,11 @@ void fillTiExecutionSummary(
                 : tipb::TIFLASH_HASH_TABLE_SIZE_KIND_BUILD_ROW_COUNT);
         hash_table_stats->set_memory_bytes(current.hash_table_stats->memory_bytes);
     }
+    // Every join kind (cross joins included) reports joined_rows once its probe finishes (0
+    // means it matched nothing) while other executors never set it, so consumers can tell
+    // joins apart by field presence instead of the value.
+    if (current.joined_rows)
+        execution_summary->mutable_tiflash_join_stats()->set_joined_rows(*current.joined_rows);
     if (current.columnar_scan_context)
         execution_summary->mutable_columnar_scan_context()->CopyFrom(current.columnar_scan_context->serialize());
     else

dbms/src/Flash/Statistics/JoinImpl.cpp

@@ -48,6 +48,7 @@ void JoinStatistics::collectExtraRuntimeDetail()
         is_spill_enabled = join_execute_info.join_profile_info->is_spill_enabled;
         is_spilled = join_execute_info.join_profile_info->is_spilled;
         hash_table_stats = join_execute_info.join_profile_info->getHashTableStats();
+        joined_rows = join_execute_info.join_profile_info->getJoinedRows();
         switch (dag_context.getExecutionMode())
         {
         case ExecutionMode::None:
@@ -71,6 +72,8 @@ void JoinStatistics::fillExtraExecutionSummary(ExecutionSummary & summary) const
 {
     if (hash_table_stats)
         summary.hash_table_stats = hash_table_stats;
+    if (joined_rows)
+        summary.joined_rows = joined_rows;
 }
 
 JoinStatistics::JoinStatistics(const tipb::Executor * executor, DAGContext & dag_context_)

dbms/src/Flash/Statistics/JoinImpl.h

@@ -46,6 +46,7 @@ class JoinStatistics : public JoinStatisticsBase
     bool is_spill_enabled = false;
     bool is_spilled = false;
     std::optional<HashTableStats> hash_table_stats;
+    std::optional<UInt64> joined_rows;
 
     BaseRuntimeStatistics join_build_base;