feat: Add reduction_factor metric to AggregateExec for EXPLAIN ANALYZE

datafusion/core/tests/sql/explain_analyze.rs

@@ -68,6 +68,13 @@ async fn explain_analyze_baseline_metrics() {
         "AggregateExec: mode=Partial, gby=[]",
         "output_bytes="
     );
+
+    assert_metrics!(
+        &formatted,
+        "AggregateExec: mode=Partial, gby=[c1@0 as c1]",
+        "reduction_factor=5.1% (5/99)"
+    );
+
     assert_metrics!(
         &formatted,
         "AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1]",

datafusion/physical-plan/src/aggregates/row_hash.rs

@@ -433,6 +433,9 @@ pub(crate) struct GroupedHashAggregateStream {
 
     /// Aggregation-specific metrics
     group_by_metrics: GroupByMetrics,
+
+    /// Reduction factor metric, calculated as `output_rows/input_rows` (only for partial aggregation)
+    reduction_factor: Option<metrics::RatioMetrics>,
 }
 
 impl GroupedHashAggregateStream {
@@ -600,6 +603,16 @@ impl GroupedHashAggregateStream {
             None
         };
 
+        let reduction_factor = if agg.mode == AggregateMode::Partial {
+            Some(
+                MetricBuilder::new(&agg.metrics)
+                    .with_type(metrics::MetricType::SUMMARY)
+                    .ratio_metrics("reduction_factor", partition),
+            )
+        } else {
+            None
+        };
+
         Ok(GroupedHashAggregateStream {
             schema: agg_schema,
             input,
@@ -620,6 +633,7 @@ impl GroupedHashAggregateStream {
             spill_state,
             group_values_soft_limit: agg.limit,
             skip_aggregation_probe,
+            reduction_factor,
         })
     }
 }
@@ -662,6 +676,11 @@ impl Stream for GroupedHashAggregateStream {
                             let timer = elapsed_compute.timer();
                             let input_rows = batch.num_rows();
 
+                            if let Some(reduction_factor) = self.reduction_factor.as_ref()
+                            {
+                                reduction_factor.add_total(input_rows);
+                            }
+
                             // Do the grouping
                             self.group_aggregate_batch(batch)?;
 
@@ -799,6 +818,11 @@ impl Stream for GroupedHashAggregateStream {
                         let output = batch.slice(0, size);
                         (ExecutionState::ProducingOutput(remaining), output)
                     };
+
+                    if let Some(reduction_factor) = self.reduction_factor.as_ref() {
+                        reduction_factor.add_part(output_batch.num_rows());
+                    }
+
                     // Empty record batches should not be emitted.
                     // They need to be treated as  [`Option<RecordBatch>`]es and handled separately
                     debug_assert!(output_batch.num_rows() > 0);