From ed306cb03c94f4acd4025557c7bf9925c73a150f Mon Sep 17 00:00:00 2001 From: LiaCastaneda Date: Fri, 2 Oct 2026 17:40:10 +0200 Subject: [PATCH] Add skewness metric to nodes that execute in partitioned mode --- datafusion/core/tests/sql/explain_analyze.rs | 78 +++++++++++++++++++ .../physical-expr-common/src/metrics/mod.rs | 16 ++++ .../physical-plan/src/aggregates/mod.rs | 2 +- .../physical-plan/src/joins/hash_join/exec.rs | 2 +- .../physical-plan/src/repartition/mod.rs | 2 +- .../src/windows/bounded_window_agg_exec.rs | 2 +- .../src/windows/window_agg_exec.rs | 2 +- .../test_files/dynamic_row_group_pruning.slt | 2 +- datafusion/sqllogictest/test_files/joins.slt | 2 +- 9 files changed, 101 insertions(+), 7 deletions(-) diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs index 67614cfbdea9a..3efe084baea35 100644 --- a/datafusion/core/tests/sql/explain_analyze.rs +++ b/datafusion/core/tests/sql/explain_analyze.rs @@ -19,6 +19,7 @@ use super::*; use insta::assert_snapshot; use rstest::rstest; +use datafusion::catalog::MemTable; use datafusion::config::ConfigOptions; use datafusion::physical_plan::display::DisplayableExecutionPlan; use datafusion::physical_plan::metrics::Timestamp; @@ -908,6 +909,83 @@ async fn parquet_explain_analyze() { ); } +/// Operators that run once per partition report `output_rows_skew`. A single +/// join/group/window key hashes every row to one partition, so each +/// hash-partitioned operator must report the maximum skew. +#[tokio::test] +#[cfg_attr(coverage, ignore)] +async fn explain_analyze_output_rows_skew_on_hash_partitioned_operators() { + let mut config = SessionConfig::new().with_target_partitions(4); + // Force a partitioned (hash repartitioned) join instead of `CollectLeft` + let options = config.options_mut(); + options.optimizer.hash_join_single_partition_threshold = 0; + options.optimizer.hash_join_single_partition_threshold_rows = 0; + let ctx = SessionContext::new_with_config(config); + // Spread the rows over 4 input partitions so they get hash repartitioned + // (a single input partition already satisfies the hash distribution) + let schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Int64, false), + Field::new("v", DataType::Int64, false), + ])); + let partitions = (0..4) + .map(|p| { + let v = (p * 25..(p + 1) * 25).collect::>(); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from(vec![1; v.len()])), + Arc::new(Int64Array::from(v)), + ], + ) + .unwrap(); + vec![batch] + }) + .collect(); + let table = MemTable::try_new(schema, partitions).unwrap(); + ctx.register_table("t", Arc::new(table)).unwrap(); + + let sql = "EXPLAIN ANALYZE \ + SELECT a.k, count(*) FROM t a JOIN t b ON a.k = b.k GROUP BY a.k"; + let formatted = explain_analyze_output(&ctx, sql).await; + for operator in [ + "RepartitionExec: partitioning=Hash", + "HashJoinExec: mode=Partitioned", + "AggregateExec: mode=SinglePartitioned", + ] { + assert_operator_metric(&formatted, operator, "output_rows_skew=100%"); + } + + let sql = "EXPLAIN ANALYZE \ + SELECT k, \ + row_number() OVER (PARTITION BY k ORDER BY v), \ + sum(v) OVER (PARTITION BY k) \ + FROM t"; + let formatted = explain_analyze_output(&ctx, sql).await; + for operator in ["BoundedWindowAggExec", "WindowAggExec"] { + assert_operator_metric(&formatted, operator, "output_rows_skew=100%"); + } +} + +async fn explain_analyze_output(ctx: &SessionContext, sql: &str) -> String { + let actual = execute_to_batches(ctx, sql).await; + arrow::util::pretty::pretty_format_batches(&actual) + .unwrap() + .to_string() +} + +fn assert_operator_metric(formatted: &str, operator: &str, metric: &str) { + // Leading space so `WindowAggExec` does not match `BoundedWindowAggExec` + let needle = format!(" {operator}"); + let line = formatted + .lines() + .find(|line| line.contains(&needle)) + .unwrap_or_else(|| panic!("{operator} not found in plan:\n{formatted}")); + assert!( + line.contains(metric), + "expected {metric} on {operator}, got:\n{line}" + ); +} + /// The Parquet scan's `elapsed_compute` must cover decoding, which dominates a /// full scan, and not only the projection of batches that are already decoded. /// See . diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index 9a1c17872e292..262550f594b19 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/datafusion/physical-expr-common/src/metrics/mod.rs @@ -250,6 +250,22 @@ impl MetricsSet { self.metrics.push(metric) } + /// Returns this set with the derived `output_rows_skew` metric appended, + /// or unchanged if no partition reported `output_rows` yet. + /// + /// This is typically used in `ExecutionPlan::metrics` of operators that + /// execute in multiple partitions (e.g. repartitions, partitioned joins and + /// aggregations), where it shows how evenly their output rows are spread + /// across partitions. With a single partition the skew is always `0%`. + /// + /// See [`BaselineMetrics::output_rows_skew_metric`] for how skew is computed. + pub fn with_output_rows_skew(mut self) -> Self { + if let Some(output_rows_skew) = BaselineMetrics::output_rows_skew_metric(&self) { + self.push(output_rows_skew); + } + self + } + /// Return the metrics whose partition ID equals `partition`. /// /// Unpartitioned metrics are excluded; an unknown partition yields an empty diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 1bf7d0c2fbfae..19d41d3232bc0 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2600,7 +2600,7 @@ impl ExecutionPlan for AggregateExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + Some(self.metrics.clone_inner().with_output_rows_skew()) } fn child_stats_requests(&self, partition: Option) -> Vec { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index bd8103db0fd1e..03ca609fa0d37 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -1856,7 +1856,7 @@ impl ExecutionPlan for HashJoinExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + Some(self.metrics.clone_inner().with_output_rows_skew()) } fn child_stats_requests(&self, partition: Option) -> Vec { diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs index 3df29b577cd1a..51ce1345fc225 100644 --- a/datafusion/physical-plan/src/repartition/mod.rs +++ b/datafusion/physical-plan/src/repartition/mod.rs @@ -1884,7 +1884,7 @@ impl ExecutionPlan for RepartitionExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + Some(self.metrics.clone_inner().with_output_rows_skew()) } fn child_stats_requests(&self, _partition: Option) -> Vec { diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs index 0c6e77a1d6f70..724a722898ff5 100644 --- a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs +++ b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs @@ -528,7 +528,7 @@ impl ExecutionPlan for BoundedWindowAggExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + Some(self.metrics.clone_inner().with_output_rows_skew()) } fn child_stats_requests(&self, partition: Option) -> Vec { diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs index dba4c063d1911..e00aaa21bb3b8 100644 --- a/datafusion/physical-plan/src/windows/window_agg_exec.rs +++ b/datafusion/physical-plan/src/windows/window_agg_exec.rs @@ -320,7 +320,7 @@ impl ExecutionPlan for WindowAggExec { } fn metrics(&self) -> Option { - Some(self.metrics.clone_inner()) + Some(self.metrics.clone_inner().with_output_rows_skew()) } fn child_stats_requests(&self, partition: Option) -> Vec { diff --git a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt index 753bf6fdfd263..10d25d7235e85 100644 --- a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt +++ b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt @@ -338,7 +338,7 @@ query TT explain analyze select rgsel.b from dim join rgsel on dim.column1 = rgsel.b; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(column1@0, b@0)], projection=[b@1], metrics=[output_rows=3, elapsed_compute=, output_bytes=, output_batches=1, build_mem_used=, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=2, input_rows=3, build_time=, join_time=, avg_fanout=100% (3/3), probe_hit_rate=100% (3/3)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(column1@0, b@0)], projection=[b@1], metrics=[output_rows=3, elapsed_compute=, output_bytes=, output_batches=1, build_mem_used=, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=2, input_rows=3, build_time=, join_time=, avg_fanout=100% (3/3), output_rows_skew=0%, probe_hit_rate=100% (3/3)] 02)--DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] 03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/rgsel.parquet]]}, projection=[b], file_type=parquet, predicate=DynamicFilter [ b@1 >= 1000 AND b@1 <= 3050 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= 1000 AND b_null_count@1 != row_count@2 AND b_min@3 <= 3050 AND b_null_count@1 != row_count@2 AND KEY_RANGE_BITMAP(b_min@3, b_max@0, 3/4096 buckets), required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, output_batches=2, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 2 matched, row_groups_pruned_bloom_filter=2 total → 2 matched, page_index_pages_pruned=20 total → 3 matched, page_index_rows_pruned=200 total → 30 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_processed=, bytes_scanned=, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=3, pushdown_rows_pruned=27, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=200, predicate_cache_records=13, bloom_filter_eval_time=, metadata_load_time=, page_index_eval_time=, row_pushdown_eval_time=, statistics_eval_time=, time_elapsed_opening=, time_elapsed_processing=, time_elapsed_scanning_total=, time_elapsed_scanning_until_data=, output_rows_skew=, scan_efficiency_ratio=] diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt index f4dbd212eb8ee..e922233fd8d55 100644 --- a/datafusion/sqllogictest/test_files/joins.slt +++ b/datafusion/sqllogictest/test_files/joins.slt @@ -5263,7 +5263,7 @@ LEFT ANTI JOIN ( ) t2 ON t1.k = t2.k; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(k@0, k@0)], metrics=[output_rows=2, elapsed_compute=, output_bytes=, output_batches=1, build_mem_used=, array_map_created_count=0, build_input_batches=0, build_input_rows=0, input_batches=1, input_rows=2, build_time=, join_time=, avg_fanout=N/A (0/0), probe_hit_rate=0% (0/2)] +01)HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(k@0, k@0)], metrics=[output_rows=2, elapsed_compute=, output_bytes=, output_batches=1, build_mem_used=, array_map_created_count=0, build_input_batches=0, build_input_rows=0, input_batches=1, input_rows=2, build_time=, join_time=, avg_fanout=N/A (0/0), output_rows_skew=0%, probe_hit_rate=0% (0/2)] 02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_rows=0, elapsed_compute=, output_bytes=, output_batches=0, expr_0_eval_time=] 03)----FilterExec: column1@0 != 1, metrics=[output_rows=0, elapsed_compute=, output_bytes=, output_batches=0, selectivity=0% (0/1)] 04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[]