diff --git a/contrib/tipb b/contrib/tipb
index fed7bc47c39..7adb3bb014c 160000
--- a/contrib/tipb
+++ b/contrib/tipb
@@ -1 +1 @@
-Subproject commit fed7bc47c39d42d21d9febf9a02291d381b7696d
+Subproject commit 7adb3bb014c41cbd73f49721dea93a953eff9ecc
diff --git a/dbms/src/Flash/Coprocessor/DAGContext.h b/dbms/src/Flash/Coprocessor/DAGContext.h
index e8a8ae5e678..26f61801f96 100644
--- a/dbms/src/Flash/Coprocessor/DAGContext.h
+++ b/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;
 };
diff --git a/dbms/src/Flash/Coprocessor/ExecutionSummary.cpp b/dbms/src/Flash/Coprocessor/ExecutionSummary.cpp
index 1e99b10c023..2374f87b1ad 100644
--- a/dbms/src/Flash/Coprocessor/ExecutionSummary.cpp
+++ b/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());
diff --git a/dbms/src/Flash/Coprocessor/ExecutionSummary.h b/dbms/src/Flash/Coprocessor/ExecutionSummary.h
index 12037dfa701..41f35ab1d05 100644
--- a/dbms/src/Flash/Coprocessor/ExecutionSummary.h
+++ b/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;
diff --git a/dbms/src/Flash/Statistics/ExecutionSummaryHelper.cpp b/dbms/src/Flash/Statistics/ExecutionSummaryHelper.cpp
index f9cdff68a52..e469e36abb9 100644
--- a/dbms/src/Flash/Statistics/ExecutionSummaryHelper.cpp
+++ b/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
diff --git a/dbms/src/Flash/Statistics/JoinImpl.cpp b/dbms/src/Flash/Statistics/JoinImpl.cpp
index cfbf6cad156..1b8821e74f0 100644
--- a/dbms/src/Flash/Statistics/JoinImpl.cpp
+++ b/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_)
diff --git a/dbms/src/Flash/Statistics/JoinImpl.h b/dbms/src/Flash/Statistics/JoinImpl.h
index f8a58d48bed..05de2b4eda3 100644
--- a/dbms/src/Flash/Statistics/JoinImpl.h
+++ b/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;
 
diff --git a/dbms/src/Flash/tests/gtest_execution_summary.cpp b/dbms/src/Flash/tests/gtest_execution_summary.cpp
index a3ab3f942dc..b4905d379c9 100644
--- a/dbms/src/Flash/tests/gtest_execution_summary.cpp
+++ b/dbms/src/Flash/tests/gtest_execution_summary.cpp
@@ -95,6 +95,9 @@ class ExecutionSummaryTestRunner : public DB::tests::ExecutorTest
             enable_join_v2 ? tipb::TIFLASH_HASH_TABLE_SIZE_KIND_BUILD_ROW_COUNT
                            : tipb::TIFLASH_HASH_TABLE_SIZE_KIND_DISTINCT_KEY_COUNT);
         ASSERT_GT(join_summary->tiflash_hash_table_stats().memory_bytes(), 0);
+        // 8 non-null probe rows match the single build-side entry which holds 8 build rows.
+        ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+        ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), 64);
     }
 
     void testHashTableStatsForMultipleJoins(bool enable_join_v2)
@@ -117,6 +120,7 @@ class ExecutionSummaryTestRunner : public DB::tests::ExecutorTest
             return true;
         });
         ASSERT_EQ(join_executor_ids.size(), 2);
+        std::vector<UInt64> joined_rows_per_join;
 
         DAGContext dag_context(*request, "test_execution_summary", concurrency);
         executeStreams(&dag_context);
@@ -145,6 +149,230 @@ class ExecutionSummaryTestRunner : public DB::tests::ExecutorTest
                 enable_join_v2 ? tipb::TIFLASH_HASH_TABLE_SIZE_KIND_BUILD_ROW_COUNT
                                : tipb::TIFLASH_HASH_TABLE_SIZE_KIND_DISTINCT_KEY_COUNT);
             ASSERT_GT(join_summary->tiflash_hash_table_stats().memory_bytes(), 0);
+            ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+            joined_rows_per_join.push_back(join_summary->tiflash_join_stats().joined_rows());
+        }
+        // The first join matches 8 probe rows against 8 build rows, the second join probes the
+        // 64 output rows of the first join against 8 build rows.
+        ASSERT_EQ(joined_rows_per_join.size(), 2);
+        std::sort(joined_rows_per_join.begin(), joined_rows_per_join.end());
+        ASSERT_EQ(joined_rows_per_join[0], 64);
+        ASSERT_EQ(joined_rows_per_join[1], 512);
+    }
+
+    void testSemiJoinHashTableStats(bool enable_join_v2)
+    {
+        context.context->getSettingsRef().enable_hash_join_v2 = enable_join_v2;
+        enablePipeline(enable_join_v2);
+
+        auto t1 = context.scan("test_db", "test_table");
+        auto t2 = context.scan("test_db", "test_table");
+        auto request = t1.join(t2, tipb::JoinType::TypeSemiJoin, {col("s1")}).build(context);
+        request->set_collect_execution_summaries(true);
+
+        DAGContext dag_context(*request, "test_execution_summary", concurrency);
+        executeStreams(&dag_context);
+        ExecutorStatisticsCollector statistics_collector("test_execution_summary", true);
+        statistics_collector.initialize(&dag_context);
+        statistics_collector.setLocalRUConsumption(
+            RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0});
+        const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries();
+
+        const tipb::ExecutorExecutionSummary * join_summary = nullptr;
+        for (const auto & summary : summaries)
+        {
+            if (summary.has_executor_id() && summary.executor_id() == "Join_2")
+            {
+                join_summary = &summary;
+                break;
+            }
+        }
+        ASSERT_NE(join_summary, nullptr);
+        ASSERT_TRUE(join_summary->has_tiflash_hash_table_stats());
+        // Semi join without other conditions stops at the first match of each probe row, so
+        // exactly one pair is enumerated for each of the 8 matched probe rows.
+        ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+        ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), 8);
+    }
+
+    void testZeroMatchJoinHashTableStats(bool enable_join_v2)
+    {
+        context.context->getSettingsRef().enable_hash_join_v2 = enable_join_v2;
+        enablePipeline(enable_join_v2);
+
+        auto request = context.scan("test_db", "test_table")
+                           .join(context.scan("test_db", "empty_table"), tipb::JoinType::TypeInnerJoin, {col("s1")})
+                           .build(context);
+        request->set_collect_execution_summaries(true);
+
+        DAGContext dag_context(*request, "test_execution_summary", concurrency);
+        executeStreams(&dag_context);
+        ExecutorStatisticsCollector statistics_collector("test_execution_summary", true);
+        statistics_collector.initialize(&dag_context);
+        statistics_collector.setLocalRUConsumption(
+            RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0});
+        const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries();
+
+        const tipb::ExecutorExecutionSummary * join_summary = nullptr;
+        for (const auto & summary : summaries)
+        {
+            if (summary.has_executor_id() && summary.executor_id() == "Join_2")
+            {
+                join_summary = &summary;
+                break;
+            }
+        }
+        ASSERT_NE(join_summary, nullptr);
+        ASSERT_TRUE(join_summary->has_tiflash_hash_table_stats());
+        ASSERT_EQ(join_summary->tiflash_hash_table_stats().size(), 0);
+        // A completed join always reports joined_rows, so "matched 0 rows" (present, 0) is
+        // distinguishable from executors that never set the field.
+        ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+        ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), 0);
+    }
+
+    void testSemiJoinWithOtherConditionStats(bool enable_join_v2)
+    {
+        context.context->getSettingsRef().enable_hash_join_v2 = enable_join_v2;
+        enablePipeline(enable_join_v2);
+
+        context.addMockTable(
+            {"test_db", "semi_join_left"},
+            {{"k", TiDB::TP::TypeLong}, {"lv", TiDB::TP::TypeLong}},
+            {toNullableVec<Int32>("k", {1, 2}), toNullableVec<Int32>("lv", {50, 50})});
+
+        const auto check_semi_join_stats
+            = [&](const std::vector<std::optional<Int64>> & rv_values, UInt64 expect_joined_rows) {
+                  // All build rows share the join key 1, so the probe row with k = 1 matches a single
+                  // entry holding rv_values.size() build rows, which the other condition `rv < lv`
+                  // checks in paced and short-circuited rounds. The probe row with k = 2 matches
+                  // nothing and must not contribute to joined_rows.
+                  context.addMockTable(
+                      {"test_db", "semi_join_right"},
+                      {{"k", TiDB::TP::TypeLong}, {"rv", TiDB::TP::TypeLong}},
+                      {toNullableVec<Int32>("k", std::vector<std::optional<Int64>>(rv_values.size(), 1)),
+                       toNullableVec<Int32>("rv", rv_values)});
+
+                  auto request = context.scan("test_db", "semi_join_left")
+                                     .join(
+                                         context.scan("test_db", "semi_join_right"),
+                                         tipb::JoinType::TypeSemiJoin,
+                                         {col("k")},
+                                         {},
+                                         {},
+                                         {lt(col("rv"), col("lv"))},
+                                         {},
+                                         0)
+                                     .build(context);
+                  request->set_collect_execution_summaries(true);
+
+                  DAGContext dag_context(*request, "test_execution_summary", concurrency);
+                  executeStreams(&dag_context);
+                  ExecutorStatisticsCollector statistics_collector("test_execution_summary", true);
+                  statistics_collector.initialize(&dag_context);
+                  statistics_collector.setLocalRUConsumption(
+                      RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0});
+                  const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries();
+
+                  const tipb::ExecutorExecutionSummary * join_summary = nullptr;
+                  for (const auto & summary : summaries)
+                  {
+                      if (summary.has_executor_id() && summary.executor_id() == "Join_2")
+                      {
+                          join_summary = &summary;
+                          break;
+                      }
+                  }
+                  ASSERT_NE(join_summary, nullptr);
+                  ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+                  ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), expect_joined_rows);
+              };
+
+        {
+            // The other condition is satisfied by the second build row, so the paced walk
+            // finishes the probe row within its first round and the remaining build rows under
+            // the entry are never enumerated: only the materialized rows are counted, not the
+            // full entry length (8) that the probe row matches. JoinV1 materializes in rounds
+            // of 2 (SemiJoinResult::pace), JoinV2 in rounds of 4 (SemiJoinProbe INITIAL_PACE).
+            check_semi_join_stats({100, 1, 2, 3, 4, 5, 6, 7}, enable_join_v2 ? 4 : 2);
+        }
+        {
+            // The other condition is never satisfied, so every build row under the entry is
+            // enumerated and counted.
+            check_semi_join_stats({100, 101, 102, 103, 104, 105, 106, 107}, 8);
+        }
+    }
+
+    void testNullAwareAntiSemiJoinWithOtherConditionStats()
+    {
+        // Null-aware semi joins are only implemented by JoinV1.
+        context.context->getSettingsRef().enable_hash_join_v2 = false;
+        enablePipeline(true);
+
+        context.addMockTable(
+            {"test_db", "semi_join_left"},
+            {{"k", TiDB::TP::TypeLong}, {"lv", TiDB::TP::TypeLong}},
+            {toNullableVec<Int32>("k", {1, 2}), toNullableVec<Int32>("lv", {50, 50})});
+
+        const auto check_na_anti_semi_join_stats
+            = [&](const std::vector<std::optional<Int64>> & rv_values, UInt64 expect_joined_rows) {
+                  // All build rows share the join key 1, so the probe row with k = 1 matches a single
+                  // entry holding rv_values.size() build rows, which the other condition `rv < lv`
+                  // checks in paced and short-circuited rounds. The probe row with k = 2 matches
+                  // nothing and must not contribute to joined_rows.
+                  context.addMockTable(
+                      {"test_db", "semi_join_right"},
+                      {{"k", TiDB::TP::TypeLong}, {"rv", TiDB::TP::TypeLong}},
+                      {toNullableVec<Int32>("k", std::vector<std::optional<Int64>>(rv_values.size(), 1)),
+                       toNullableVec<Int32>("rv", rv_values)});
+
+                  auto request = context.scan("test_db", "semi_join_left")
+                                     .join(
+                                         context.scan("test_db", "semi_join_right"),
+                                         tipb::JoinType::TypeAntiSemiJoin,
+                                         {col("k")},
+                                         {},
+                                         {},
+                                         {lt(col("rv"), col("lv"))},
+                                         {},
+                                         0,
+                                         /*is_null_aware_semi_join=*/true)
+                                     .build(context);
+                  request->set_collect_execution_summaries(true);
+
+                  DAGContext dag_context(*request, "test_execution_summary", concurrency);
+                  executeStreams(&dag_context);
+                  ExecutorStatisticsCollector statistics_collector("test_execution_summary", true);
+                  statistics_collector.initialize(&dag_context);
+                  statistics_collector.setLocalRUConsumption(
+                      RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0});
+                  const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries();
+
+                  const tipb::ExecutorExecutionSummary * join_summary = nullptr;
+                  for (const auto & summary : summaries)
+                  {
+                      if (summary.has_executor_id() && summary.executor_id() == "Join_2")
+                      {
+                          join_summary = &summary;
+                          break;
+                      }
+                  }
+                  ASSERT_NE(join_summary, nullptr);
+                  ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+                  ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), expect_joined_rows);
+              };
+
+        {
+            // The other condition is satisfied by the second build row, so the paced walk
+            // finishes the probe row within its first round (NASemiJoinResult::pace is 2) and
+            // the remaining build rows under the entry are never enumerated: only the
+            // materialized rows are counted, not the full entry length (8).
+            check_na_anti_semi_join_stats({100, 1, 2, 3, 4, 5, 6, 7}, 2);
+        }
+        {
+            // The other condition is never satisfied, so every build row under the entry is
+            // enumerated and counted.
+            check_na_anti_semi_join_stats({100, 101, 102, 103, 104, 105, 106, 107}, 8);
         }
     }
 
@@ -199,6 +427,72 @@ class ExecutionSummaryTestRunner : public DB::tests::ExecutorTest
         // Ten build workers own independent maps, so the reported total is not the final merged NDV.
         ASSERT_GE(aggregation_summary->tiflash_hash_table_stats().size(), 3);
         ASSERT_GT(aggregation_summary->tiflash_hash_table_stats().memory_bytes(), 0);
+        // joined_rows is a join-only field; hash aggregation never reports it.
+        ASSERT_FALSE(aggregation_summary->has_tiflash_join_stats());
+    }
+
+    void testCrossJoinStats(bool enable_pipeline, size_t shallow_copy_cross_probe_threshold)
+    {
+        enablePipeline(enable_pipeline);
+        const auto old_threshold = context.context->getSettingsRef().shallow_copy_cross_probe_threshold;
+        context.context->setSetting(
+            "shallow_copy_cross_probe_threshold",
+            Field(static_cast<UInt64>(shallow_copy_cross_probe_threshold)));
+
+        // Cross joins have no hash table but still report the candidate pairs enumerated by
+        // their probes. Empty join keys make the mock executor build a cross join.
+        const auto check_cross_join_stats
+            = [&](const std::shared_ptr<tipb::DAGRequest> & request, UInt64 expect_joined_rows) {
+                  request->set_collect_execution_summaries(true);
+
+                  DAGContext dag_context(*request, "test_execution_summary", concurrency);
+                  executeStreams(&dag_context);
+                  ExecutorStatisticsCollector statistics_collector("test_execution_summary", true);
+                  statistics_collector.initialize(&dag_context);
+                  statistics_collector.setLocalRUConsumption(
+                      RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0});
+                  const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries();
+
+                  const tipb::ExecutorExecutionSummary * join_summary = nullptr;
+                  for (const auto & summary : summaries)
+                  {
+                      if (summary.has_executor_id() && summary.executor_id() == "Join_2")
+                      {
+                          join_summary = &summary;
+                          break;
+                      }
+                  }
+                  ASSERT_NE(join_summary, nullptr);
+                  ASSERT_FALSE(join_summary->has_tiflash_hash_table_stats());
+                  ASSERT_TRUE(join_summary->has_tiflash_join_stats());
+                  ASSERT_EQ(join_summary->tiflash_join_stats().joined_rows(), expect_joined_rows);
+              };
+
+        {
+            // Cross inner join: every probe row is paired with all 12 build rows.
+            auto request = context.scan("test_db", "test_table")
+                               .join(context.scan("test_db", "test_table"), tipb::JoinType::TypeInnerJoin, {})
+                               .build(context);
+            check_cross_join_stats(request, 144);
+        }
+        {
+            // Cross semi join (Any strictness) stops at the first match of each probe row, so
+            // exactly one pair is enumerated per probe row.
+            auto request = context.scan("test_db", "test_table")
+                               .join(context.scan("test_db", "test_table"), tipb::JoinType::TypeSemiJoin, {})
+                               .build(context);
+            check_cross_join_stats(request, 12);
+        }
+        {
+            // A cross join whose build side is empty enumerates no pair, but still reports the
+            // present-but-zero count once its probe finishes.
+            auto request = context.scan("test_db", "test_table")
+                               .join(context.scan("test_db", "empty_table"), tipb::JoinType::TypeInnerJoin, {})
+                               .build(context);
+            check_cross_join_stats(request, 0);
+        }
+
+        context.context->setSetting("shallow_copy_cross_probe_threshold", Field(static_cast<UInt64>(old_threshold)));
     }
 
 #define WRAP_FOR_EXCUTION_SUMMARY_TEST_BEGIN                                      \
@@ -312,6 +606,37 @@ try
 }
 CATCH
 
+TEST_F(ExecutionSummaryTestRunner, semiJoinHashTableStats)
+try
+{
+    testSemiJoinHashTableStats(false);
+    testSemiJoinHashTableStats(true);
+}
+CATCH
+
+TEST_F(ExecutionSummaryTestRunner, zeroMatchJoinHashTableStats)
+try
+{
+    testZeroMatchJoinHashTableStats(false);
+    testZeroMatchJoinHashTableStats(true);
+}
+CATCH
+
+TEST_F(ExecutionSummaryTestRunner, semiJoinWithOtherConditionStats)
+try
+{
+    testSemiJoinWithOtherConditionStats(false);
+    testSemiJoinWithOtherConditionStats(true);
+}
+CATCH
+
+TEST_F(ExecutionSummaryTestRunner, nullAwareAntiSemiJoinWithOtherConditionStats)
+try
+{
+    testNullAwareAntiSemiJoinWithOtherConditionStats();
+}
+CATCH
+
 TEST_F(ExecutionSummaryTestRunner, hashAggTableStats)
 try
 {
@@ -321,6 +646,19 @@ try
 }
 CATCH
 
+TEST_F(ExecutionSummaryTestRunner, crossJoinStats)
+try
+{
+    for (auto enable_pipeline : {false, true})
+    {
+        // The first threshold forces the shallow copy cross probe mode, the second one keeps
+        // the deep copy mode.
+        testCrossJoinStats(enable_pipeline, 1);
+        testCrossJoinStats(enable_pipeline, 1024);
+    }
+}
+CATCH
+
 TEST_F(ExecutionSummaryTestRunner, genMPPTaskExecutionInfoWithoutSetRUInfo)
 try
 {
diff --git a/dbms/src/Flash/tests/gtest_spill_join.cpp b/dbms/src/Flash/tests/gtest_spill_join.cpp
index 1a585ec5b4f..e904234cba9 100644
--- a/dbms/src/Flash/tests/gtest_spill_join.cpp
+++ b/dbms/src/Flash/tests/gtest_spill_join.cpp
@@ -316,6 +316,12 @@ try
     ASSERT_EQ(hash_table_stats->size, 10);
     ASSERT_EQ(hash_table_stats->size_kind, HashTableSizeKind::DistinctKeyCount);
     ASSERT_GT(hash_table_stats->memory_bytes, 0);
+    // Every probe key matches exactly 3 build rows, so the 10 probe rows produce 30 matched pairs
+    // in total. The count stays complete across spilling because each restore join contributes
+    // the pairs of the partitions it re-probes.
+    const auto joined_rows = join_execute_info.join_profile_info->getJoinedRows();
+    ASSERT_TRUE(joined_rows.has_value());
+    ASSERT_EQ(*joined_rows, 30);
 }
 CATCH
 
diff --git a/dbms/src/Interpreters/CrossJoinProbeHelper.cpp b/dbms/src/Interpreters/CrossJoinProbeHelper.cpp
index 8c464432963..122ff6fcd1b 100644
--- a/dbms/src/Interpreters/CrossJoinProbeHelper.cpp
+++ b/dbms/src/Interpreters/CrossJoinProbeHelper.cpp
@@ -442,6 +442,7 @@ Block crossProbeBlockDeepCopyRightBlockImpl(ProbeProcessInfo & probe_process_inf
     auto * filter_ptr = probe_process_info.filter.get();
     auto * offset_ptr = probe_process_info.offsets_to_replicate.get();
 
+    UInt64 joined_rows_in_call = 0;
     for (; current_row < block_rows; ++current_row)
     {
         if constexpr (has_row_filter_map)
@@ -479,6 +480,14 @@ Block crossProbeBlockDeepCopyRightBlockImpl(ProbeProcessInfo & probe_process_inf
                 offset_ptr,
                 probe_process_info.cross_join_data->right_rows_to_be_added_when_matched,
                 probe_process_info.max_block_size);
+            if (!block_full)
+            {
+                /// Count the candidate pairs of this probe row before other conditions are
+                /// applied: all right rows for All strictness, the single first-match pair for
+                /// Any strictness (right_rows_to_be_added_when_matched is already capped to 1
+                /// there, which also matches the first-match counting of hash semi/anti joins).
+                joined_rows_in_call += probe_process_info.cross_join_data->right_rows_to_be_added_when_matched;
+            }
         }
         else
         {
@@ -498,6 +507,8 @@ Block crossProbeBlockDeepCopyRightBlockImpl(ProbeProcessInfo & probe_process_inf
             break;
     }
     probe_process_info.updateEndRow<false>(current_row);
+    if (joined_rows_in_call != 0 && probe_process_info.joined_rows_counter)
+        probe_process_info.joined_rows_counter->fetch_add(joined_rows_in_call, std::memory_order_relaxed);
     return probe_process_info.cross_join_data->result_block_schema.cloneWithColumns(std::move(dst_columns));
 }
 
@@ -633,6 +644,11 @@ std::pair<Block, bool> crossProbeBlockShallowCopyRightBlockImpl(
     (*probe_process_info.offsets_to_replicate)[0] = right_row;
     probe_process_info.cross_join_data->next_right_block_index++;
     probe_process_info.updateEndRow<true>(probe_process_info.start_row + 1);
+    /// Count the candidate pairs of this probe row against the current right block before other
+    /// conditions are applied. Shallow mode only serves All strictness (Any is capped to one
+    /// pair and stays in deep mode), so every call materializes all rows of the right block.
+    if (probe_process_info.joined_rows_counter)
+        probe_process_info.joined_rows_counter->fetch_add(right_row, std::memory_order_relaxed);
     return {block, true};
 }
 } // namespace
diff --git a/dbms/src/Interpreters/Join.cpp b/dbms/src/Interpreters/Join.cpp
index e1fa575e285..1607d4734b3 100644
--- a/dbms/src/Interpreters/Join.cpp
+++ b/dbms/src/Interpreters/Join.cpp
@@ -1653,7 +1653,7 @@ Block doNASemiJoinOrSemiJoin(
     {
         if (is_cancelled())
             return {};
-        helper->doJoin();
+        helper->doJoin(probe_process_info);
         // because pipeline model is enabled by default, current_task_timer should be not null in most senarios
         if likely (current_task_timer != nullptr)
         {
@@ -2051,6 +2051,9 @@ void Join::finalizeProfileInfo()
     // TODO: Aggregate restore join profile fields, such as peak build memory usage, when exposing
     // them in execution summaries.
     finalizeHashTableStats();
+    // Every join kind (cross joins included) reports the pairs enumerated by its own probe
+    // streams exactly once; each restore join contributes the partitions it re-probes.
+    profile_info->addJoinedRows(joined_rows.load(std::memory_order_relaxed));
 }
 
 void Join::finalizeHashTableStats()
@@ -2224,14 +2227,22 @@ Block Join::joinBlock(ProbeProcessInfo & probe_process_info) const
     Block block{};
 
     using enum ASTTableJoin::Kind;
+    /// Bind the accumulator on every call so a reused ProbeProcessInfo always counts into
+    /// the Join that is actually probing (e.g. main join and its restore joins).
+    probe_process_info.joined_rows_counter = &joined_rows;
     if (isCrossJoin(kind))
+    {
         block = joinBlockCross(probe_process_info);
-    else if (isNullAwareSemiFamily(kind))
-        block = joinBlockNullAwareSemi(probe_process_info);
-    else if (isSemiFamily(kind) || isLeftOuterSemiFamily(kind))
-        block = joinBlockSemi(probe_process_info);
+    }
     else
-        block = joinBlockHash(probe_process_info);
+    {
+        if (isNullAwareSemiFamily(kind))
+            block = joinBlockNullAwareSemi(probe_process_info);
+        else if (isSemiFamily(kind) || isLeftOuterSemiFamily(kind))
+            block = joinBlockSemi(probe_process_info);
+        else
+            block = joinBlockHash(probe_process_info);
+    }
 
     // if cancelled, just return empty block
     if (!block)
diff --git a/dbms/src/Interpreters/Join.h b/dbms/src/Interpreters/Join.h
index 2509546478b..c1846ebdbb6 100644
--- a/dbms/src/Interpreters/Join.h
+++ b/dbms/src/Interpreters/Join.h
@@ -396,6 +396,10 @@ class Join
     JoinProfileInfoPtr profile_info = std::make_shared<JoinProfileInfo>();
     /// Keep this flag per Join object so every completed restore hash table is merged exactly once.
     bool hash_table_stats_finalized = false;
+    /// Total number of (probe row, build row) pairs matched by the equi-join keys, accumulated by all
+    /// concurrent probe streams of this join and reported via profile_info when the last stream
+    /// finishes. Restore joins count their own re-probing and merge into the same profile.
+    mutable std::atomic<UInt64> joined_rows{0};
     HashJoinSpillContextPtr hash_join_spill_context;
     const Block & getOutputBlock() const { return finalized ? output_block_after_finalize : output_block; }
     const Names & getRequiredColumns() const { return required_columns; }
diff --git a/dbms/src/Interpreters/JoinPartition.cpp b/dbms/src/Interpreters/JoinPartition.cpp
index 622c87b92a9..ee40bd207b1 100644
--- a/dbms/src/Interpreters/JoinPartition.cpp
+++ b/dbms/src/Interpreters/JoinPartition.cpp
@@ -1543,6 +1543,9 @@ void NO_INLINE probeBlockImplTypeCase(
     const auto & build_hash_data = probe_process_info.hash_join_data->hash_data->getData();
     size_t i;
     bool block_full = false;
+    /// Matched (probe row, build row) pairs found by this call; flushed into the Join-owned
+    /// accumulator once below to avoid per-row atomic operations.
+    UInt64 joined_rows_in_call = 0;
     for (i = probe_process_info.start_row; i < rows; ++i)
     {
         if (has_row_filter_map && (*row_filter_map)[i])
@@ -1626,6 +1629,7 @@ void NO_INLINE probeBlockImplTypeCase(
             auto it = internal_map.find(key, hash_value);
             if (it != internal_map.end())
             {
+                const auto & matched_value = static_cast<const typename Map::mapped_type::Base_t &>(it->getMapped());
                 if constexpr (row_flagged_map)
                 {
                     block_full = RowFlaggedHashMapAdder<Map>::addFound(
@@ -1637,11 +1641,18 @@ void NO_INLINE probeBlockImplTypeCase(
                         offsets_to_replicate.get(),
                         right_indexes,
                         probe_process_info);
+                    /// addFound returns true without inserting anything when the block is full and
+                    /// the probe row will be retried, so only count after a successful insert.
+                    if (!block_full)
+                        joined_rows_in_call += matched_value.list_length;
                 }
                 else if constexpr (KIND == ASTTableJoin::Kind::RightSemi || KIND == ASTTableJoin::Kind::RightAnti)
                 {
                     /// For RightSemi/RightAnti without other conditions, just flag the hash entry is enough
                     it->getMapped().setUsed();
+                    /// No rows are inserted for this probe row, but every build row under the matched
+                    /// entry is still a matched pair of it.
+                    joined_rows_in_call += matched_value.list_length;
                 }
                 else
                 {
@@ -1658,6 +1669,9 @@ void NO_INLINE probeBlockImplTypeCase(
                         offsets_to_replicate.get(),
                         right_indexes,
                         probe_process_info);
+                    /// Same as above: only count after a successful insert.
+                    if (!block_full)
+                        joined_rows_in_call += matched_value.list_length;
                 }
             }
             else
@@ -1701,6 +1715,9 @@ void NO_INLINE probeBlockImplTypeCase(
         }
     }
 
+    if (joined_rows_in_call != 0 && probe_process_info.joined_rows_counter)
+        probe_process_info.joined_rows_counter->fetch_add(joined_rows_in_call, std::memory_order_relaxed);
+
     probe_process_info.updateEndRow<false>(i);
 }
 
@@ -1765,7 +1782,8 @@ probeBlockNullAwareSemiInternal(
     const Sizes & key_sizes,
     const TiDB::TiDBCollators & collators,
     const NALeftSideInfo & left_side_info,
-    const NARightSideInfo & right_side_info)
+    const NARightSideInfo & right_side_info,
+    std::atomic<UInt64> * joined_rows_counter)
 {
     static_assert(
         KIND == ASTTableJoin::Kind::NullAware_Anti || KIND == ASTTableJoin::Kind::NullAware_LeftOuterAnti
@@ -1779,6 +1797,10 @@ probeBlockNullAwareSemiInternal(
     Arena pool;
     size_t segment_size = join_partitions.size();
 
+    /// Matched (probe row, build row) pairs found by this call; flushed into the Join-owned
+    /// accumulator once below to avoid per-row atomic operations.
+    UInt64 joined_rows_in_call = 0;
+
     PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>> res;
     res.reserve(rows);
     std::list<NASemiJoinResult<KIND, STRICTNESS> *> res_list;
@@ -1861,11 +1883,17 @@ probeBlockNullAwareSemiInternal(
                 /// E.g. (1,2) in ((1,2),(1,3),(null,3))
                 res.emplace_back(i, NASemiJoinStep::DONE, nullptr);
                 res.back().template setResult<SemiJoinResultType::TRUE_VALUE>();
+                /// Any null-aware semi stops at the first match, so exactly one pair is enumerated
+                /// for each matched probe row.
+                joined_rows_in_call += 1;
             }
             else
             {
                 /// Else the other condition must be checked for these matched right row(s).
                 auto map_it = &static_cast<const typename Map::mapped_type::Base_t &>(it->getMapped());
+                /// All strictness checks other conditions for these matched right row(s) later, in
+                /// paced and short-circuited rounds driven by NASemiJoinHelper::doJoin(); the pairs
+                /// actually enumerated are counted there, not the full entry length here.
                 res.emplace_back(i, NASemiJoinStep::NOT_NULL_KEY_CHECK_MATCHED_ROWS, static_cast<const void *>(map_it));
                 res_list.emplace_back(&res.back());
             }
@@ -1901,6 +1929,8 @@ probeBlockNullAwareSemiInternal(
         res.emplace_back(i, NASemiJoinStep::NOT_NULL_KEY_CHECK_NULL_ROWS, nullptr);
         res_list.push_back(&res.back());
     }
+    if (joined_rows_in_call != 0 && joined_rows_counter)
+        joined_rows_counter->fetch_add(joined_rows_in_call, std::memory_order_relaxed);
     return std::make_pair(std::move(res), std::move(res_list));
 }
 
@@ -1912,7 +1942,8 @@ std::pair<PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>>, std::list<NASemiJo
     const Sizes & key_sizes,
     const TiDB::TiDBCollators & collators,
     const NALeftSideInfo & left_side_info,
-    const NARightSideInfo & right_side_info)
+    const NARightSideInfo & right_side_info,
+    std::atomic<UInt64> * joined_rows_counter)
 {
 #define CALL(has_key_null_map, has_row_filter_map)                                                                  \
     return probeBlockNullAwareSemiInternal<KIND, STRICTNESS, KeyGetter, Map, has_key_null_map, has_row_filter_map>( \
@@ -1922,7 +1953,8 @@ std::pair<PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>>, std::list<NASemiJo
         key_sizes,                                                                                                  \
         collators,                                                                                                  \
         left_side_info,                                                                                             \
-        right_side_info);
+        right_side_info,                                                                                            \
+        joined_rows_counter);
 
     if (left_side_info.key_null_map)
     {
@@ -1990,6 +2022,9 @@ probeBlockSemiInternal(
     sort_key_containers.resize(probe_process_info.hash_join_data->key_columns.size());
     Arena pool;
     bool need_virtual_dispatch_for_probe_block = join_build_info.needVirtualDispatchForProbeBlock();
+    /// Matched (probe row, build row) pairs found by this call; flushed into the Join-owned
+    /// accumulator once below to avoid per-row atomic operations.
+    UInt64 joined_rows_in_call = 0;
     if (need_virtual_dispatch_for_probe_block)
     {
         RUNTIME_ASSERT(!(join_build_info.restore_round > 0 && join_build_info.enable_fine_grained_shuffle));
@@ -2078,10 +2113,16 @@ probeBlockSemiInternal(
             if constexpr (STRICTNESS == ASTTableJoin::Strictness::Any)
             {
                 res.emplace_back(true);
+                /// Any semi/anti stops at the first match, so exactly one pair is enumerated for
+                /// each matched probe row.
+                joined_rows_in_call += 1;
             }
             else
             {
                 auto map_it = &static_cast<const typename Map::mapped_type::Base_t &>(it->getMapped());
+                /// All strictness checks other conditions for these matched right row(s) later, in
+                /// paced and short-circuited rounds driven by SemiJoinHelper::doJoin(); the pairs
+                /// actually enumerated are counted there, not the full entry length here.
                 res.emplace_back(i, static_cast<const void *>(map_it));
                 res_list.emplace_back(&res.back());
             }
@@ -2103,6 +2144,8 @@ probeBlockSemiInternal(
     {
         RUNTIME_CHECK_MSG(res_list.empty(), "SemiJoinResult list must be empty if strictness is any");
     }
+    if (joined_rows_in_call != 0 && probe_process_info.joined_rows_counter)
+        probe_process_info.joined_rows_counter->fetch_add(joined_rows_in_call, std::memory_order_relaxed);
     return std::make_pair(std::move(res), std::move(res_list));
 }
 
@@ -2262,7 +2305,8 @@ std::pair<PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>>, std::list<NASemiJo
         const Sizes & key_sizes,
         const TiDB::TiDBCollators & collators,
         const NALeftSideInfo & left_side_info,
-        const NARightSideInfo & right_side_info)
+        const NARightSideInfo & right_side_info,
+        std::atomic<UInt64> * joined_rows_counter)
 {
     auto method = join_partitions[0]->join_map_method;
     switch (method)
@@ -2280,7 +2324,8 @@ std::pair<PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>>, std::list<NASemiJo
             key_sizes,                                                                           \
             collators,                                                                           \
             left_side_info,                                                                      \
-            right_side_info);
+            right_side_info,                                                                     \
+            joined_rows_counter);
         APPLY_FOR_JOIN_VARIANTS(M)
 #undef M
 
@@ -2299,7 +2344,8 @@ std::pair<PaddedPODArray<NASemiJoinResult<KIND, STRICTNESS>>, std::list<NASemiJo
             const Sizes & key_sizes,                                                                                \
             const TiDB::TiDBCollators & collators,                                                                  \
             const NALeftSideInfo & left_side_info,                                                                  \
-            const NARightSideInfo & right_side_info);
+            const NARightSideInfo & right_side_info,                                                                \
+            std::atomic<UInt64> * joined_rows_counter);
 
 APPLY_FOR_NULL_AWARE_SEMI_JOIN(M)
 #undef M
diff --git a/dbms/src/Interpreters/JoinPartition.h b/dbms/src/Interpreters/JoinPartition.h
index 6dad6e13e70..274662ad021 100644
--- a/dbms/src/Interpreters/JoinPartition.h
+++ b/dbms/src/Interpreters/JoinPartition.h
@@ -221,7 +221,8 @@ class JoinPartition
         const Sizes & key_sizes,
         const TiDB::TiDBCollators & collators,
         const NALeftSideInfo & left_side_info,
-        const NARightSideInfo & right_side_info);
+        const NARightSideInfo & right_side_info,
+        std::atomic<UInt64> * joined_rows_counter);
 
     template <ASTTableJoin::Kind KIND, ASTTableJoin::Strictness STRICTNESS, typename Maps>
     static std::pair<PaddedPODArray<SemiJoinResult<KIND, STRICTNESS>>, std::list<SemiJoinResult<KIND, STRICTNESS> *>> probeBlockSemi(
diff --git a/dbms/src/Interpreters/JoinV2/HashJoin.cpp b/dbms/src/Interpreters/JoinV2/HashJoin.cpp
index e93787ff9a7..bb7e36a6432 100644
--- a/dbms/src/Interpreters/JoinV2/HashJoin.cpp
+++ b/dbms/src/Interpreters/JoinV2/HashJoin.cpp
@@ -444,17 +444,31 @@ bool HashJoin::finishOneProbe(size_t stream_index)
     auto & wd = probe_workers_data[stream_index];
     LOG_DEBUG(
         log,
-        "{} probe handle {} rows, cost {}ms(hash_table {}ms + replicate {}ms + other condition {}ms), collision {}",
+        "{} probe handle {} rows, cost {}ms(hash_table {}ms + replicate {}ms + other condition {}ms), collision {}, "
+        "joined rows {}",
         stream_index,
         wd.probe_handle_rows,
         wd.probe_time / 1000000UL,
         wd.probe_hash_table_time / 1000000UL,
         wd.replicate_time / 1000000UL,
         wd.other_condition_time / 1000000UL,
-        wd.collision);
+        wd.collision,
+        wd.joined_rows);
     if (active_probe_worker.fetch_sub(1) == 1)
     {
         FAIL_POINT_TRIGGER_EXCEPTION(FailPoints::exception_mpp_hash_probe);
+        // The last finishing worker sees all other workers' counts through the fetch_sub chain, so
+        // the sum is complete. Cross join does not use JoinV2 today; the guard is defensive: the
+        // cross probe path has no key-comparison loop, so wd.joined_rows would stay zero and
+        // publishing the sum would report a misleading present-0 instead of leaving the
+        // statistic absent.
+        if (method != HashJoinKeyMethod::Cross)
+        {
+            size_t total_joined_rows = 0;
+            for (const auto & probe_wd : probe_workers_data)
+                total_joined_rows += probe_wd.joined_rows;
+            profile_info->addJoinedRows(total_joined_rows);
+        }
         return true;
     }
     return false;
diff --git a/dbms/src/Interpreters/JoinV2/HashJoinProbe.cpp b/dbms/src/Interpreters/JoinV2/HashJoinProbe.cpp
index b3c71247b9e..f02b4695282 100644
--- a/dbms/src/Interpreters/JoinV2/HashJoinProbe.cpp
+++ b/dbms/src/Interpreters/JoinV2/HashJoinProbe.cpp
@@ -664,6 +664,9 @@ void JoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDa
     RowPtr ptr = ctx.current_build_row_ptr;
     bool is_matched = ctx.current_row_is_matched;
     size_t collision = 0;
+    /// Each build row in the pointer-table chain is compared exactly once (resumed states keep
+    /// the mid-chain position), so every successful comparison is one matched pair.
+    size_t joined_rows = 0;
     constexpr size_t key_offset
         = sizeof(RowPtr) + (KeyGetterType::joinKeyCompareHashFirst() ? sizeof(HashValueType) : 0);
 
@@ -725,6 +728,7 @@ void JoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDa
             collision += !key_is_equal;
             if (key_is_equal)
             {
+                ++joined_rows;
                 if constexpr (Adder::need_not_matched)
                     is_matched = true;
 
@@ -775,6 +779,7 @@ void JoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDa
     ctx.current_build_row_ptr = ptr;
     ctx.current_row_is_matched = is_matched;
     wd.collision += collision;
+    wd.joined_rows += joined_rows;
 
 #undef NOT_MATCHED
 }
@@ -843,6 +848,9 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
     if constexpr (has_other_condition)
         RUNTIME_CHECK(current_offset == 0);
     size_t collision = 0;
+    /// Each build row in the pointer-table chain is compared exactly once (prefetch states walk
+    /// one chain node per step), so every successful comparison is one matched pair.
+    size_t joined_rows = 0;
     constexpr size_t key_offset
         = sizeof(RowPtr) + (KeyGetterType::joinKeyCompareHashFirst() ? sizeof(HashValueType) : 0);
 
@@ -874,6 +882,7 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
                 state->is_matched |= key_is_equal;
             if (key_is_equal)
             {
+                ++joined_rows;
                 if constexpr (Adder::need_matched)
                 {
                     bool is_end = Adder::addMatched(
@@ -1005,6 +1014,7 @@ void JoinProbeHelper::probeFillColumnsPrefetch(
     ctx.prefetch_active_states = active_states;
     ctx.prefetch_iter = k;
     wd.collision += collision;
+    wd.joined_rows += joined_rows;
 
 #undef NOT_MATCHED
 }
diff --git a/dbms/src/Interpreters/JoinV2/HashJoinProbe.h b/dbms/src/Interpreters/JoinV2/HashJoinProbe.h
index 698f1dba409..b9fec8cf2e4 100644
--- a/dbms/src/Interpreters/JoinV2/HashJoinProbe.h
+++ b/dbms/src/Interpreters/JoinV2/HashJoinProbe.h
@@ -113,6 +113,9 @@ struct alignas(CPU_CACHE_LINE_SIZE) JoinProbeWorkerData
     size_t replicate_time = 0;
     size_t other_condition_time = 0;
     size_t collision = 0;
+    /// Matched (probe row, build row) pairs enumerated by this worker; summed into the join
+    /// profile by the last finishing probe worker.
+    size_t joined_rows = 0;
 };
 
 class JoinProbeHelperUtil
diff --git a/dbms/src/Interpreters/JoinV2/SemiJoinProbe.cpp b/dbms/src/Interpreters/JoinV2/SemiJoinProbe.cpp
index 02f9cb8f797..d534645e334 100644
--- a/dbms/src/Interpreters/JoinV2/SemiJoinProbe.cpp
+++ b/dbms/src/Interpreters/JoinV2/SemiJoinProbe.cpp
@@ -261,6 +261,9 @@ SemiJoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDat
     RUNTIME_CHECK(probe_list->slotCapacity() == ctx.rows);
     size_t current_offset = 0;
     size_t collision = 0;
+    /// Each build row in the pointer-table chain is compared exactly once (paced walks resume via
+    /// probe_row.build_row_ptr), so every successful comparison is one matched pair.
+    size_t joined_rows = 0;
     size_t key_offset = sizeof(RowPtr);
     if constexpr (KeyGetterType::joinKeyCompareHashFirst())
     {
@@ -272,6 +275,7 @@ SemiJoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDat
         fillNullMapWithZero<false>(added_columns);
 
         wd.collision += collision;
+        wd.joined_rows += joined_rows;
     });
 
     auto iter_end = probe_list->end();
@@ -288,6 +292,7 @@ SemiJoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDat
             bool key_is_equal = joinKeyIsEqual(key_getter, probe_row.key, key2, probe_row.hash, ptr);
             collision += !key_is_equal;
             current_offset += key_is_equal;
+            joined_rows += key_is_equal;
             if (key_is_equal)
             {
                 wd.selective_offsets.push_back(idx);
@@ -359,6 +364,7 @@ SemiJoinProbeHelper::probeFillColumns(JoinProbeContext & ctx, JoinProbeWorkerDat
             bool key_is_equal = joinKeyIsEqual(key_getter, key, key2, hash, ptr);
             collision += !key_is_equal;
             current_offset += key_is_equal;
+            joined_rows += key_is_equal;
             if (key_is_equal)
             {
                 wd.selective_offsets.push_back(idx);
@@ -461,6 +467,9 @@ void NO_INLINE SemiJoinProbeHelper::probeFillColumnsPrefetch(
     size_t k = ctx.prefetch_iter;
     size_t current_offset = 0;
     size_t collision = 0;
+    /// Each build row in the pointer-table chain is compared exactly once (prefetch states walk
+    /// one chain node per step), so every successful comparison is one matched pair.
+    size_t joined_rows = 0;
     constexpr size_t key_offset
         = sizeof(RowPtr) + (KeyGetterType::joinKeyCompareHashFirst() ? sizeof(HashValueType) : 0);
 
@@ -483,6 +492,7 @@ void NO_INLINE SemiJoinProbeHelper::probeFillColumnsPrefetch(
             state->is_matched |= key_is_equal;
             current_offset += key_is_equal;
             state->remaining_pace -= key_is_equal;
+            joined_rows += key_is_equal;
             bool remaining_pace_is_zero = false;
             if (key_is_equal)
             {
@@ -656,6 +666,7 @@ void NO_INLINE SemiJoinProbeHelper::probeFillColumnsPrefetch(
     ctx.prefetch_active_states = active_states;
     ctx.prefetch_iter = k;
     wd.collision += collision;
+    wd.joined_rows += joined_rows;
 }
 
 template <typename KeyGetter, ASTTableJoin::Kind kind>
diff --git a/dbms/src/Interpreters/NullAwareSemiJoinHelper.cpp b/dbms/src/Interpreters/NullAwareSemiJoinHelper.cpp
index a53b5e34df9..14c7bcca86c 100644
--- a/dbms/src/Interpreters/NullAwareSemiJoinHelper.cpp
+++ b/dbms/src/Interpreters/NullAwareSemiJoinHelper.cpp
@@ -343,7 +343,8 @@ void NASemiJoinHelper<KIND, STRICTNESS, Maps>::probeHashTable(
         key_sizes,
         collators,
         left_side_info,
-        right_side_info);
+        right_side_info,
+        probe_process_info.joined_rows_counter);
 
     RUNTIME_ASSERT(
         join_result.size() == probe_rows,
@@ -401,7 +402,7 @@ void NASemiJoinHelper<KIND, STRICTNESS, Maps>::probeHashTable(
 }
 
 template <ASTTableJoin::Kind KIND, ASTTableJoin::Strictness STRICTNESS, typename Maps>
-void NASemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin()
+void NASemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin([[maybe_unused]] ProbeProcessInfo & probe_process_info)
 {
     switch (current_check_step)
     {
@@ -410,7 +411,13 @@ void NASemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin()
         // if STRICTNESS is Any, we add this check only to avoid static_assert fail
         if constexpr (STRICTNESS == All)
         {
-            runStep<NASemiJoinStep::NOT_NULL_KEY_CHECK_MATCHED_ROWS>();
+            const size_t joined_rows_in_round = runStep<NASemiJoinStep::NOT_NULL_KEY_CHECK_MATCHED_ROWS>();
+            /// Count the pairs actually enumerated this round: other conditions are evaluated on
+            /// the materialized rows, and a satisfied condition finishes the probe row early (the
+            /// paced walk short-circuits), so count the materialized rows instead of the full
+            /// matched entry length.
+            if (joined_rows_in_round != 0 && probe_process_info.joined_rows_counter != nullptr)
+                probe_process_info.joined_rows_counter->fetch_add(joined_rows_in_round, std::memory_order_relaxed);
             if (undetermined_result_list.empty())
             {
                 // go to next step
@@ -482,14 +489,14 @@ void NASemiJoinHelper<KIND, STRICTNESS, Maps>::prepareForRunStep()
 
 template <ASTTableJoin::Kind KIND, ASTTableJoin::Strictness STRICTNESS, typename Maps>
 template <NASemiJoinStep STEP>
-void NASemiJoinHelper<KIND, STRICTNESS, Maps>::runStep()
+size_t NASemiJoinHelper<KIND, STRICTNESS, Maps>::runStep()
 {
     static_assert(
         STEP == NASemiJoinStep::NOT_NULL_KEY_CHECK_MATCHED_ROWS || STEP == NASemiJoinStep::NOT_NULL_KEY_CHECK_NULL_ROWS
         || STEP == NASemiJoinStep::NULL_KEY_CHECK_NULL_ROWS);
 
     if (undetermined_result_list.empty())
-        return;
+        return 0;
 
     std::vector<size_t> offsets;
     size_t block_columns = res_block.columns();
@@ -544,6 +551,7 @@ void NASemiJoinHelper<KIND, STRICTNESS, Maps>::runStep()
     exec_block = res_block.cloneWithColumns(std::move(columns));
 
     runAndCheckExprResult<STEP>(exec_block, offsets);
+    return current_offset;
 }
 
 template <ASTTableJoin::Kind KIND, ASTTableJoin::Strictness STRICTNESS, typename Maps>
diff --git a/dbms/src/Interpreters/NullAwareSemiJoinHelper.h b/dbms/src/Interpreters/NullAwareSemiJoinHelper.h
index e37fbe6a305..b5fd203c852 100644
--- a/dbms/src/Interpreters/NullAwareSemiJoinHelper.h
+++ b/dbms/src/Interpreters/NullAwareSemiJoinHelper.h
@@ -184,7 +184,7 @@ class NASemiJoinHelper
         const ProbeProcessInfo & probe_process_info,
         const NameSet & probe_output_name_set,
         const Block & right_sample_block);
-    void doJoin();
+    void doJoin(ProbeProcessInfo & probe_process_info);
     bool isJoinDone() const
     {
         return is_probe_hash_table_done && undetermined_result_list.empty()
@@ -195,8 +195,10 @@ class NASemiJoinHelper
     Block genJoinResult(const NameSet & output_column_names_set);
 
 private:
+    /// Run one paced round for the given step and return the number of build rows materialized
+    /// this round (candidate pairs actually enumerated).
     template <NASemiJoinStep STEP>
-    void runStep();
+    size_t runStep();
     template <NASemiJoinStep STEP>
     void prepareForRunStep();
     void runStepAllBlocks();
diff --git a/dbms/src/Interpreters/ProbeProcessInfo.h b/dbms/src/Interpreters/ProbeProcessInfo.h
index a9d5c4a0d92..72ed42e927c 100644
--- a/dbms/src/Interpreters/ProbeProcessInfo.h
+++ b/dbms/src/Interpreters/ProbeProcessInfo.h
@@ -18,6 +18,8 @@
 #include <Core/Block.h>
 #include <Parsers/ASTTablesInSelectQuery.h>
 
+#include <atomic>
+
 namespace DB
 {
 enum class CrossProbeMode
@@ -126,6 +128,11 @@ struct ProbeProcessInfo
     /// for null-aware join
     std::unique_ptr<NullAwareJoinProbeProcessData> null_aware_join_data;
 
+    /// Points to the Join-owned accumulator of hash-table matched row pairs. Set by Join::joinBlock
+    /// before probing each block; probe implementations count matched pairs locally per call and
+    /// flush into it once per call with relaxed ordering. Cross joins leave it null.
+    std::atomic<UInt64> * joined_rows_counter = nullptr;
+
     // for semi join family
     std::unique_ptr<void, std::function<void(void *)>> semi_join_family_helper = nullptr; /// type erasure
 
diff --git a/dbms/src/Interpreters/SemiJoinHelper.cpp b/dbms/src/Interpreters/SemiJoinHelper.cpp
index 0727e9d9664..14e7031025b 100644
--- a/dbms/src/Interpreters/SemiJoinHelper.cpp
+++ b/dbms/src/Interpreters/SemiJoinHelper.cpp
@@ -156,7 +156,7 @@ SemiJoinHelper<KIND, STRICTNESS, Maps>::SemiJoinHelper(
 }
 
 template <ASTTableJoin::Kind KIND, ASTTableJoin::Strictness STRICTNESS, typename Maps>
-void SemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin()
+void SemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin([[maybe_unused]] ProbeProcessInfo & probe_process_info)
 {
     if constexpr (STRICTNESS == ASTTableJoin::Strictness::Any)
     {
@@ -210,6 +210,13 @@ void SemiJoinHelper<KIND, STRICTNESS, Maps>::doJoin()
                 break;
         }
 
+        /// The pairs actually enumerated this round: other conditions are evaluated on the
+        /// materialized rows, and a satisfied condition finishes the probe row early (the paced
+        /// walk short-circuits), so count the materialized rows instead of the full matched
+        /// entry length.
+        if (current_offset != 0 && probe_process_info.joined_rows_counter != nullptr)
+            probe_process_info.joined_rows_counter->fetch_add(current_offset, std::memory_order_relaxed);
+
         /// Move the columns to exec_block.
         /// Note that this can remove the new columns that are added in the last loop
         /// from equal and other condition expressions.
diff --git a/dbms/src/Interpreters/SemiJoinHelper.h b/dbms/src/Interpreters/SemiJoinHelper.h
index b168dc496fe..60d06944c4f 100644
--- a/dbms/src/Interpreters/SemiJoinHelper.h
+++ b/dbms/src/Interpreters/SemiJoinHelper.h
@@ -145,7 +145,7 @@ class SemiJoinHelper
         const ProbeProcessInfo & probe_process_info,
         const NameSet & probe_output_name_set,
         const Block & right_sample_block);
-    void doJoin();
+    void doJoin(ProbeProcessInfo & probe_process_info);
 
     bool isJoinDone() const { return is_probe_hash_table_done && undetermined_result_list.empty(); }
     bool isProbeHashTableDone() const { return is_probe_hash_table_done; }
diff --git a/tests/fullstack-test/expr/generated_index.test b/tests/fullstack-test/expr/generated_index.test
index 45ffe96ef25..d132875fd6e 100644
--- a/tests/fullstack-test/expr/generated_index.test
+++ b/tests/fullstack-test/expr/generated_index.test
@@ -17,6 +17,12 @@ mysql> create table test.t(c1 varchar(100), c2 varchar(100));
 mysql> insert into test.t values('ABC', 'DEF');
 mysql> alter table test.t set tiflash replica 1;
 func> wait_table test t
+
+# NextGen rejects add index with local sort when SEM is enabled (global sort is
+# the only supported path there), so the generated index cannot be created in
+# the next-gen environment. Skip the rest of this test for next-gen runs.
+#SKIP_FOR_NEXT_GEN
+
 mysql> alter table test.t add index idx2((lower(c2)));
 
 mysql> select /*+ nth_plan(1) */ * from test.t where lower(test.t.c2) = 'def';