feat: add avg_fanout summary metric to HashJoinExec

Author: feniljain

datafusion/physical-plan/src/joins/hash_join/stream.rs

@@ -537,6 +537,12 @@ impl HashJoinStream {
             .probe_hit_rate
             .add_part(distinct_right_indices_count);
 
+        self.join_metrics.avg_fanout.add_part(left_indices.len());
+
+        self.join_metrics
+            .avg_fanout
+            .add_total(distinct_right_indices_count);
+
         // apply join filter if exists
         let (left_indices, right_indices) = if let Some(filter) = &self.filter {
             apply_join_filter_to_indices(

datafusion/physical-plan/src/joins/utils.rs

@@ -27,7 +27,9 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 
 use crate::joins::SharedBitmapBuilder;
-use crate::metrics::{self, BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricType};
+use crate::metrics::{
+    self, BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricType,
+};
 use crate::projection::{ProjectionExec, ProjectionExpr};
 use crate::{
     ColumnStatistics, ExecutionPlan, ExecutionPlanProperties, Partitioning, Statistics,
@@ -1331,6 +1333,8 @@ pub(crate) struct BuildProbeJoinMetrics {
     pub(crate) output_batches: metrics::Count,
     /// Fraction of probe rows that found more than one match
     pub(crate) probe_hit_rate: metrics::RatioMetrics,
+    /// Average number of build matches per matched probe row
+    pub(crate) avg_fanout: metrics::RatioMetrics,
 }
 
 // This Drop implementation updates the elapsed compute part of the metrics.
@@ -1381,6 +1385,10 @@ impl BuildProbeJoinMetrics {
             .with_type(MetricType::SUMMARY)
             .ratio_metrics("probe_hit_rate", partition);
 
+        let avg_fanout = MetricBuilder::new(metrics)
+            .with_type(MetricType::SUMMARY)
+            .ratio_metrics("avg_fanout", partition);
+
         Self {
             build_time,
             build_input_batches,
@@ -1392,6 +1400,7 @@ impl BuildProbeJoinMetrics {
             output_batches,
             baseline,
             probe_hit_rate,
+            avg_fanout,
         }
     }
 }