feat(metric): Add output skewness metric to detect skewed plans easier (#21211)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #.

## Rationale for this change

### Output Skewness Metric

This introduces a metric to quantify execution skew across partitions,
inspired by recent work on intra-partition parallelism for Parquet scans
(#20529).
cc @Dandandan and @alamb (I think this is related to your ongoing work)

For example, a Parquet scan with 4 partitions:
```
partition 1: output_rows = 100
partition 2: output_rows = 0
partition 3: output_rows = 0
partition 4: output_rows = 0
```
represents a highly skewed workload, where most work is concentrated in
a single partition.

This metric normalizes skew into the range `[0%, 100%]`, where `0%`
indicates a perfectly balanced distribution and `100%` indicates maximal
skew. This makes it easy to detect skew via simple thresholds in
automated tooling.

### Demo
Run clickbench q41, the parquet exec has skew score 92%, from `EXPLAIN
ANALYZE VERBOSE` we can verify only 2 out of 14 partitions has output
rows from the parquet scan.
```sh
DataFusion CLI v52.3.0
> CREATE or replace EXTERNAL TABLE hits STORED AS PARQUET LOCATION '/Users/yongting/Code/datafusion/benchmarks/data/hits_partitioned';
0 row(s) fetched.
Elapsed 0.338 seconds.

-- Clickbench Q41
> explain analyze SELECT "WindowClientWidth", "WindowClientHeight", COUNT(*) AS PageViews FROM hits WHERE "CounterID" = 62 AND "EventDate" >= '2013-07-01' AND "EventDate" <= '2013-07-31' AND "IsRefresh" = 0 AND "DontCountHits" = 0 AND "URLHash" = 2868770270353813622 GROUP BY "WindowClientWidth", "WindowClientHeight" ORDER BY PageViews DESC LIMIT 10 OFFSET 10000;
...
DataSourceExec: ..., output_rows_skew=92.35%...] |
```
---

### Definition

Let `r_i` be the output rows of partition `i`:
```
# ranges from [1, partition_count], if perfectly balanced, it reaches partition_count for maximum effective parallelism
effective_parallelism = (sum(r_i))^2 / sum(r_i^2)

# convert effective parallelism to a skewness from 0%(perfectly balanced) to 100%(extreme skew)
output_rows_skew =
  (1 - (effective_parallelism - 1) / (partition_count - 1)) * 100%
```

Examples:
- `[10, 10, 10, 10]` → skew = `0%`
- `[40, 0, 0, 0]` → skew = `100%`

---

### Motivation

I think there are several more further works to do to tackle the
skewness, so this metric can help. Even with intra-partition parallelism
(e.g., morsel-driven Parquet scan), skew can still propagate to
downstream operators:
```
partition 1: parquet_scan(output_rows=100) → FilterExec(90)
partition 2: parquet_scan(0)   → FilterExec(0)
partition 3: parquet_scan(0)   → FilterExec(0)
partition 4: parquet_scan(0)   → FilterExec(0)
```
Downstream operators like `FilterExec` may remain underutilized,
suggesting the need for additional techniques such as internal
parallelism or repartitioning. This metric helps identify such cases and
guide further optimization.

---

### Why calculate on output_rows
It's easy to implement since `output_rows` are already tracked, the
downside is it requires some knowledge to interpret, for instance if we
see a large skewness value on parquet DataSource, we want to ensure a)
parquet has internal parallelism mechanism to evenly distribute the work
b) the downstream won't be affected by the produced skewness

## What changes are included in this PR?

Add a derived `output_rows_skew` to DataSourceExec with parquet source,
in the follow-ups I think it's also useful to other datasources, and
FilterExec (which might also introduce skewness during execution)

## Are these changes tested?

<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
sqllogictests

## Are there any user-facing changes?
No
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->

<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->

Author: 2010YOUY01

datafusion/core/tests/sql/explain_analyze.rs

@@ -69,7 +69,7 @@ async fn explain_analyze_baseline_metrics() {
     assert_metrics!(
         &formatted,
         "AggregateExec: mode=Partial, gby=[c1@0 as c1]",
-        "reduction_factor=5.1% (5/99)"
+        "reduction_factor=5.05% (5/99)"
     );
 
     {
@@ -887,7 +887,8 @@ async fn parquet_explain_analyze() {
         &formatted,
         "row_groups_pruned_statistics=1 total \u{2192} 1 matched"
     );
-    assert_contains!(&formatted, "scan_efficiency_ratio=14%");
+    assert_contains!(&formatted, "output_rows_skew=0%");
+    assert_contains!(&formatted, "scan_efficiency_ratio=13.99%");
 
     // The order of metrics is expected to be the same as the actual pruning order
     // (file-> row-group -> page)

datafusion/datasource/src/source.rs

@@ -27,7 +27,9 @@ use datafusion_physical_plan::execution_plan::{
     Boundedness, EmissionType, SchedulingType,
 };
 use datafusion_physical_plan::metrics::SplitMetrics;
-use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
+use datafusion_physical_plan::metrics::{
+    BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet,
+};
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::stream::BatchSplitStream;
 use datafusion_physical_plan::{
@@ -356,7 +358,20 @@ impl ExecutionPlan for DataSourceExec {
     }
 
     fn metrics(&self) -> Option<MetricsSet> {
-        Some(self.data_source.metrics().clone_inner())
+        let mut metrics = self.data_source.metrics().clone_inner();
+
+        // Add `output_rows_skew` metric to the metrics set.
+        // Done here because it's a derived metric from output_rows metric.
+        if let Some(file_scan_config) =
+            self.data_source.as_any().downcast_ref::<FileScanConfig>()
+            && file_scan_config.file_source().file_type() == "parquet"
+            && let Some(output_rows_skew) =
+                BaselineMetrics::output_rows_skew_metric(&metrics)
+        {
+            metrics.push(output_rows_skew);
+        }
+
+        Some(metrics)
     }
 
     fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {

datafusion/physical-expr-common/src/metrics/baseline.rs

@@ -17,12 +17,16 @@
 
 //! Metrics common for almost all operators
 
-use std::task::Poll;
+use std::{borrow::Cow, collections::BTreeMap, sync::Arc, task::Poll};
 
 use arrow::record_batch::RecordBatch;
 use datafusion_common::{Result, utils::memory::get_record_batch_memory_size};
 
-use super::{Count, ExecutionPlanMetricsSet, MetricBuilder, Time, Timestamp};
+use super::{
+    Count, ExecutionPlanMetricsSet, Metric, MetricBuilder, MetricsSet, Time, Timestamp,
+};
+
+const OUTPUT_ROWS_SKEW_METRIC_NAME: &str = "output_rows_skew";
 
 /// Helper for creating and tracking common "baseline" metrics for
 /// each operator
@@ -125,6 +129,61 @@ impl BaselineMetrics {
         &self.output_batches
     }
 
+    /// Returns a derived metric that summarizes how unevenly `output_rows`
+    /// are distributed across partitions.
+    ///
+    /// The score is normalized to the range `[0%, 100%]`, where `0%`
+    /// indicates a perfectly balanced distribution and `100%` indicates the
+    /// most skewed distribution.
+    ///
+    /// The calculation is:
+    /// `effective_parallelism = square(sum(r_i)) / sum(square(r_i))`
+    /// `output_rows_skew = (1 - ((effective_parallelism - 1) / (partition_count - 1))) * 100%`
+    ///
+    /// Example: for 4 partitions with output rows `[10, 10, 10, 10]`,
+    /// `effective_parallelism = 40^2 / (10^2 + 10^2 + 10^2 + 10^2) = 4`,
+    /// so `output_rows_skew = 0%`. For `[40, 0, 0, 0]`, the score is `100%`.
+    pub fn output_rows_skew_metric(metrics: &MetricsSet) -> Option<Arc<Metric>> {
+        let output_rows = metrics
+            .iter()
+            .filter_map(|metric| match (metric.partition(), metric.value()) {
+                (Some(partition), super::MetricValue::OutputRows(count)) => {
+                    Some((partition, count.value() as u128))
+                }
+                _ => None,
+            })
+            .fold(
+                BTreeMap::<usize, u128>::new(),
+                |mut output_rows, (partition, rows)| {
+                    *output_rows.entry(partition).or_default() += rows;
+                    output_rows
+                },
+            )
+            .into_values()
+            .collect::<Vec<_>>();
+
+        if output_rows.is_empty() {
+            return None;
+        }
+
+        let ratio_metrics = super::RatioMetrics::new().with_display_raw_values(false);
+        if let Some(score) = output_rows_skew_score(&output_rows) {
+            ratio_metrics.set_part((score * 10_000.0).round() as usize);
+            ratio_metrics.set_total(10_000);
+        }
+
+        Some(Arc::new(
+            Metric::new(
+                super::MetricValue::Ratio {
+                    name: Cow::Borrowed(OUTPUT_ROWS_SKEW_METRIC_NAME),
+                    ratio_metrics,
+                },
+                None,
+            )
+            .with_type(super::MetricType::DEV),
+        ))
+    }
+
     /// Records the fact that this operator's execution is complete
     /// (recording the `end_time` metric).
     ///
@@ -178,6 +237,38 @@ impl Drop for BaselineMetrics {
     }
 }
 
+/// See [`BaselineMetrics::output_rows_skew_metric`] for the algorithm.
+fn output_rows_skew_score(output_rows: &[u128]) -> Option<f64> {
+    if output_rows.is_empty() {
+        return None;
+    }
+
+    let partition_count = output_rows.len();
+    if partition_count == 1 {
+        return Some(0.0);
+    }
+
+    let (total_rows, sum_of_squares) =
+        output_rows
+            .iter()
+            .fold((0.0, 0.0), |(total_rows, sum_of_squares), rows| {
+                let rows = *rows as f64;
+                (total_rows + rows, sum_of_squares + rows.powi(2))
+            });
+    if total_rows == 0.0 {
+        return None;
+    }
+
+    if sum_of_squares == 0.0 {
+        return None;
+    }
+
+    let effective_parallelism = total_rows.powi(2) / sum_of_squares;
+    let balanced_score = (effective_parallelism - 1.0) / (partition_count as f64 - 1.0);
+
+    Some((1.0 - balanced_score).clamp(0.0, 1.0))
+}
+
 /// Helper for creating and tracking spill-related metrics for
 /// each operator
 #[derive(Debug, Clone)]

datafusion/physical-expr-common/src/metrics/value.rs

@@ -468,6 +468,8 @@ pub struct RatioMetrics {
     part: Arc<AtomicUsize>,
     total: Arc<AtomicUsize>,
     merge_strategy: RatioMergeStrategy,
+    /// Ratios are displayed as `1% (1/100)`; this controls the latter part.
+    display_raw_values: bool,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -485,6 +487,7 @@ impl RatioMetrics {
             part: Arc::new(AtomicUsize::new(0)),
             total: Arc::new(AtomicUsize::new(0)),
             merge_strategy: RatioMergeStrategy::AddPartAddTotal,
+            display_raw_values: true,
         }
     }
 
@@ -493,6 +496,11 @@ impl RatioMetrics {
         self
     }
 
+    pub fn with_display_raw_values(mut self, display_raw_values: bool) -> Self {
+        self.display_raw_values = display_raw_values;
+        self
+    }
+
     /// Add `n` to the numerator (`part`) value
     pub fn add_part(&self, n: usize) {
         self.part.fetch_add(n, Ordering::Relaxed);
@@ -544,44 +552,53 @@ impl RatioMetrics {
 
 impl PartialEq for RatioMetrics {
     fn eq(&self, other: &Self) -> bool {
-        self.part() == other.part() && self.total() == other.total()
+        self.part() == other.part()
+            && self.total() == other.total()
+            && self.display_raw_values == other.display_raw_values
     }
 }
 
-/// Format a float number with `digits` most significant numbers.
-///
-/// fmt_significant(12.5) -> "12"
-/// fmt_significant(0.0543) -> "0.054"
-/// fmt_significant(0.000123) -> "0.00012"
-fn fmt_significant(mut x: f64, digits: usize) -> String {
-    if x == 0.0 {
-        return "0".to_string();
-    }
-
-    let exp = x.abs().log10().floor(); // exponent of first significant digit
-    let scale = 10f64.powf(-(exp - (digits as f64 - 1.0)));
-    x = (x * scale).round() / scale; // round to N significant digits
-    format!("{x}")
-}
-
 impl Display for RatioMetrics {
+    /// Format the ratio to a format like '18.26% (220/1150)'
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let part = self.part();
         let total = self.total();
 
+        // Format the ratio first (for example, `6667/10000` -> `66.67%`),
+        // then optionally append the raw values as ` (6.67 K/10.00 K)`.
+        if total == 0 {
+            write!(f, "N/A")?;
+        } else {
+            // Use basis points so we can round with integer math:
+            // e.g. 18.26% has basis point 1826
+            let basis_points = (((part as u128 * 10_000) + (total as u128 / 2))
+                / total as u128) as usize;
+            let whole = basis_points / 100;
+            let fractional = basis_points % 100;
+
+            if fractional == 0 {
+                write!(f, "{whole}%")?;
+            } else if fractional.is_multiple_of(10) {
+                write!(f, "{whole}.{}%", fractional / 10)?;
+            } else {
+                write!(f, "{whole}.{fractional:02}%")?;
+            }
+        }
+
+        if !self.display_raw_values {
+            return Ok(());
+        }
+
         if total == 0 {
             if part == 0 {
-                write!(f, "N/A (0/0)")
+                write!(f, " (0/0)")
             } else {
-                write!(f, "N/A ({}/0)", human_readable_count(part))
+                write!(f, " ({}/0)", human_readable_count(part))
             }
         } else {
-            let percentage = (part as f64 / total as f64) * 100.0;
-
             write!(
                 f,
-                "{}% ({}/{})",
-                fmt_significant(percentage, 2),
+                " ({}/{})",
                 human_readable_count(part),
                 human_readable_count(total)
             )
@@ -865,7 +882,8 @@ impl MetricValue {
                 Self::Ratio {
                     name: name.clone(),
                     ratio_metrics: RatioMetrics::new()
-                        .with_merge_strategy(merge_strategy),
+                        .with_merge_strategy(merge_strategy)
+                        .with_display_raw_values(ratio_metrics.display_raw_values),
                 }
             }
             Self::Custom { name, value } => Self::Custom {
@@ -1232,7 +1250,21 @@ mod tests {
         };
         tiny_ratio_metrics.add_part(1);
         tiny_ratio_metrics.add_total(3000);
-        assert_eq!("0.033% (1/3.00 K)", tiny_ratio.to_string());
+        assert_eq!("0.03% (1/3.00 K)", tiny_ratio.to_string());
+
+        ratio_metrics.set_part(6667);
+        ratio_metrics.set_total(10_000);
+        assert_eq!("66.67% (6.67 K/10.00 K)", ratio.to_string());
+
+        let percentage_only = RatioMetrics::new().with_display_raw_values(false);
+        let ratio = MetricValue::Ratio {
+            name: Cow::Borrowed("percentage_only"),
+            ratio_metrics: percentage_only.clone(),
+        };
+        assert_eq!("N/A", ratio.to_string());
+        percentage_only.set_part(6667);
+        percentage_only.set_total(10_000);
+        assert_eq!("66.67%", ratio.to_string());
     }
 
     #[test]

datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt

@@ -104,7 +104,7 @@ Plan with Metrics
 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, <slt:ignore>]
 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=<slt:ignore>, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=18% (210/1.15 K)]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=<slt:ignore>, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=18.31% (210/1.15 K)]
 
 statement ok
 set datafusion.explain.analyze_level = dev;