feat: Add per-expression evaluation timing metrics to ProjectionExec (#19447)

2010YOUY01 · web-flow · commit 36df145e9761 · 2025-12-29T14:24:47.000Z
## Which issue does this PR close?  Part of #18456 ## Rationale for this change  See issue for the rationale, this PR is implementing the 1st part of the issue. Demo in `datafrusion-cli` ``` DataFusion CLI v51.0.0 > explain analyze select a+1, pow(a,2) from generate_series(1,1000000) as t1(a); +-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | plan_type | plan | +-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | Plan with Metrics | ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=1.00 M, elapsed_compute=32.37ms, output_bytes=15.3 MB, output_batches=123, expr_eval_time_0=8.27ms, expr_eval_time_1=23.82ms] | | | RepartitionExec: partitioning=RoundRobinBatch(14), input_partitions=1, metrics=[output_rows=1.00 M, elapsed_compute=780.87µs, output_bytes=7.7 MB, output_batches=123, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, fetch_time=1.19ms, repartition_time=1ns, send_time=186.54µs] | | | ProjectionExec: expr=[value@0 as a], metrics=[output_rows=1.00 M, elapsed_compute=148.62µs, output_bytes=7.7 MB, output_batches=123, expr_eval_time_0=4.12µs] | | | LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=1000000, batch_size=8192], metrics=[output_rows=1.00 M, elapsed_compute=1.01ms, output_bytes=7.7 MB, output_batches=123] | | | | +-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ 1 row(s) fetched. ``` Each expressions' evaluation time are tracked individually in `expr_eval_time_*`. `expr_eval_time_0` is for `a+1`, and `expr_eval_time_1` is for `pow(a,2)`. I chose number index because the pretty formatting for expressions are still a bit verbose now (like `[a@0 + 1 as t1.a + Int64(1)`) ## What changes are included in this PR?  1. Moved metrics module from `datafusion-execution` crate to `datafusion-physical-expr-common`. (I optimistically think it's enough to move to `execution` crate previously, however there are many execution utilities live in `physical-expr-common`, so we have to further move it down in the dependency tree) 2. Added a struct to hold each expression's evaluation time: `ExpressionEvaluatorMetrics` 3. Put `Option<ExpressionEvaluatorMetrics>` inside `Projector`, and change the execution accordingly. ## Are these changes tested?  Yes, added several slts ## Are there any user-facing changes? No
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/datafusion/execution/src/lib.rs b/datafusion/execution/src/lib.rs
@@ -32,7 +32,6 @@ pub mod cache;
 pub mod config;
 pub mod disk_manager;
 pub mod memory_pool;
-pub mod metrics;
 pub mod object_store;
 #[cfg(feature = "parquet_encryption")]
 pub mod parquet_encryption;
diff --git a/datafusion/physical-expr-common/Cargo.toml b/datafusion/physical-expr-common/Cargo.toml
@@ -43,7 +43,9 @@ name = "datafusion_physical_expr_common"
 [dependencies]
 ahash = { workspace = true }
 arrow = { workspace = true }
+chrono = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-expr-common = { workspace = true }
 hashbrown = { workspace = true }
 itertools = { workspace = true }
+parking_lot = { workspace = true }
diff --git a/datafusion/physical-expr-common/src/lib.rs b/datafusion/physical-expr-common/src/lib.rs
@@ -35,6 +35,7 @@
 pub mod binary_map;
 pub mod binary_view_map;
 pub mod datum;
+pub mod metrics;
 pub mod physical_expr;
 pub mod sort_expr;
 pub mod tree_node;
diff --git a/datafusion/physical-expr-common/src/metrics/baseline.rs b/datafusion/physical-expr-common/src/metrics/baseline.rs
@@ -29,7 +29,9 @@ use super::{Count, ExecutionPlanMetricsSet, MetricBuilder, Time, Timestamp};
 ///
 /// Example:
 /// ```
-/// use datafusion_execution::metrics::{BaselineMetrics, ExecutionPlanMetricsSet};
+/// use datafusion_physical_expr_common::metrics::{
+///     BaselineMetrics, ExecutionPlanMetricsSet,
+/// };
 /// let metrics = ExecutionPlanMetricsSet::new();
 ///
 /// let partition = 2;
diff --git a/datafusion/physical-expr-common/src/metrics/builder.rs b/datafusion/physical-expr-common/src/metrics/builder.rs
@@ -34,7 +34,7 @@ use super::{
 /// case of constant strings
 ///
 /// ```rust
-/// use datafusion_execution::metrics::*;
+/// use datafusion_physical_expr_common::metrics::*;
 ///
 /// let metrics = ExecutionPlanMetricsSet::new();
 /// let partition = 1;
diff --git a/datafusion/physical-expr-common/src/metrics/custom.rs b/datafusion/physical-expr-common/src/metrics/custom.rs
@@ -44,7 +44,7 @@ use std::{any::Any, fmt::Debug, fmt::Display, sync::Arc};
 /// # use std::any::Any;
 /// # use std::sync::atomic::{AtomicUsize, Ordering};
 ///
-/// # use datafusion_execution::metrics::CustomMetricValue;
+/// # use datafusion_physical_expr_common::metrics::CustomMetricValue;
 ///
 /// #[derive(Debug, Default)]
 /// struct MyCounter {
diff --git a/datafusion/physical-expr-common/src/metrics/expression.rs b/datafusion/physical-expr-common/src/metrics/expression.rs
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Metrics helpers for expression evaluation.
+
+use super::{ExecutionPlanMetricsSet, MetricBuilder, MetricType, ScopedTimerGuard, Time};
+
+/// Tracks evaluation time for a sequence of expressions.
+///
+/// # Example
+/// Given SQL query:
+///     EXPLAIN ANALYZE
+///     SELECT a+1, pow(a,2)
+///     FROM generate_series(1, 1000000) as t1(a)
+///
+/// This struct holds two time metrics for the projection expressions
+/// `a+1` and `pow(a,2)`, respectively.
+///
+/// The output reads:
+/// `ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[... expr_0_eval_time=9.23ms, expr_1_eval_time=32.35ms...]`
+#[derive(Debug, Clone)]
+pub struct ExpressionEvaluatorMetrics {
+    expression_times: Vec<Time>,
+}
+
+impl ExpressionEvaluatorMetrics {
+    /// Create metrics for a collection of expressions.
+    ///
+    /// # Args
+    /// - metrics: see `MetricBuilder` for details.
+    /// - partition: see `MetricBuilder` for details.
+    /// - expression_labels: unique identifier for each metric, so that the metric
+    ///   can get aggregated across multiple partitions. It is not the name showed
+    ///   in the `EXPLAIN ANALYZE`, the metric name will be `expr_{idx}_eval_time`
+    ///   according to the expression order.
+    pub fn new<T>(
+        metrics: &ExecutionPlanMetricsSet,
+        partition: usize,
+        expression_labels: impl IntoIterator<Item = T>,
+    ) -> Self
+    where
+        T: Into<String>,
+    {
+        let expression_times = expression_labels
+            .into_iter()
+            .enumerate()
+            .map(|(idx, label)| {
+                MetricBuilder::new(metrics)
+                    .with_new_label("expr", label.into())
+                    .with_type(MetricType::DEV)
+                    // Existing PhysicalExpr formatter is a bit verbose, so use simple name
+                    .subset_time(format!("expr_{idx}_eval_time"), partition)
+            })
+            .collect();
+
+        Self { expression_times }
+    }
+
+    /// Returns a timer guard for the expression at `index`, if present.
+    #[inline]
+    pub fn scoped_timer(&self, index: usize) -> Option<ScopedTimerGuard<'_>> {
+        self.expression_times.get(index).map(Time::timer)
+    }
+
+    /// The number of tracked expressions.
+    pub fn len(&self) -> usize {
+        self.expression_times.len()
+    }
+
+    /// True when no expressions are tracked.
+    pub fn is_empty(&self) -> bool {
+        self.expression_times.is_empty()
+    }
+}
diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs
@@ -20,6 +20,7 @@
 mod baseline;
 mod builder;
 mod custom;
+mod expression;
 mod value;
 
 use datafusion_common::HashMap;
@@ -35,6 +36,7 @@ use std::{
 pub use baseline::{BaselineMetrics, RecordOutput, SpillMetrics, SplitMetrics};
 pub use builder::MetricBuilder;
 pub use custom::CustomMetricValue;
+pub use expression::ExpressionEvaluatorMetrics;
 pub use value::{
     Count, Gauge, MetricValue, PruningMetrics, RatioMergeStrategy, RatioMetrics,
     ScopedTimerGuard, Time, Timestamp,
@@ -47,7 +49,7 @@ pub use value::{
 /// [`ExecutionPlanMetricsSet`].
 ///
 /// ```
-/// use datafusion_execution::metrics::*;
+/// use datafusion_physical_expr_common::metrics::*;
 ///
 /// let metrics = ExecutionPlanMetricsSet::new();
 /// assert!(metrics.clone_inner().output_rows().is_none());
diff --git a/datafusion/physical-expr-common/src/metrics/value.rs b/datafusion/physical-expr-common/src/metrics/value.rs
diff --git a/datafusion/physical-expr-common/src/utils.rs b/datafusion/physical-expr-common/src/utils.rs
@@ -17,6 +17,7 @@
 
 use std::sync::Arc;
 
+use crate::metrics::ExpressionEvaluatorMetrics;
 use crate::physical_expr::PhysicalExpr;
 use crate::tree_node::ExprContext;
 
@@ -101,11 +102,25 @@ pub fn scatter(mask: &BooleanArray, truthy: &dyn Array) -> Result<ArrayRef> {
 pub fn evaluate_expressions_to_arrays<'a>(
     exprs: impl IntoIterator<Item = &'a Arc<dyn PhysicalExpr>>,
     batch: &RecordBatch,
+) -> Result<Vec<ArrayRef>> {
+    evaluate_expressions_to_arrays_with_metrics(exprs, batch, None)
+}
+
+/// Same as [`evaluate_expressions_to_arrays`] but records optional per-expression metrics.
+///
+/// For metrics tracking, see [`ExpressionEvaluatorMetrics`] for details.
+#[inline]
+pub fn evaluate_expressions_to_arrays_with_metrics<'a>(
+    exprs: impl IntoIterator<Item = &'a Arc<dyn PhysicalExpr>>,
+    batch: &RecordBatch,
+    metrics: Option<&ExpressionEvaluatorMetrics>,
 ) -> Result<Vec<ArrayRef>> {
     let num_rows = batch.num_rows();
     exprs
         .into_iter()
-        .map(|e| {
+        .enumerate()
+        .map(|(idx, e)| {
+            let _timer = metrics.and_then(|m| m.scoped_timer(idx));
             e.evaluate(batch)
                 .and_then(|col| col.into_array_of_size(num_rows))
         })
diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs
@@ -32,8 +32,11 @@ use datafusion_common::{
     Result, assert_or_internal_err, internal_datafusion_err, plan_err,
 };
 
+use datafusion_physical_expr_common::metrics::ExecutionPlanMetricsSet;
+use datafusion_physical_expr_common::metrics::ExpressionEvaluatorMetrics;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
-use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
+use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays_with_metrics;
 use indexmap::IndexMap;
 use itertools::Itertools;
 
@@ -481,9 +484,30 @@ impl ProjectionExprs {
         Ok(Projector {
             projection: self.clone(),
             output_schema,
+            expression_metrics: None,
         })
     }
 
+    pub fn create_expression_metrics(
+        &self,
+        metrics: &ExecutionPlanMetricsSet,
+        partition: usize,
+    ) -> ExpressionEvaluatorMetrics {
+        let labels: Vec<String> = self
+            .exprs
+            .iter()
+            .map(|proj_expr| {
+                let expr_sql = fmt_sql(proj_expr.expr.as_ref()).to_string();
+                if proj_expr.expr.to_string() == proj_expr.alias {
+                    expr_sql
+                } else {
+                    format!("{expr_sql} AS {}", proj_expr.alias)
+                }
+            })
+            .collect();
+        ExpressionEvaluatorMetrics::new(metrics, partition, labels)
+    }
+
     /// Project statistics according to this projection.
     /// For example, for a projection `SELECT a AS x, b + 1 AS y`, where `a` is at index 0 and `b` is at index 1,
     /// if the input statistics has column statistics for columns `a`, `b`, and `c`, the output statistics would have column statistics for columns `x` and `y`.
@@ -621,19 +645,41 @@ impl<'a> IntoIterator for &'a ProjectionExprs {
 pub struct Projector {
     projection: ProjectionExprs,
     output_schema: SchemaRef,
+    /// If `Some`, metrics will be tracked for projection evaluation.
+    expression_metrics: Option<ExpressionEvaluatorMetrics>,
 }
 
 impl Projector {
+    /// Construct the projector with metrics. After execution, related metrics will
+    /// be tracked inside `ExecutionPlanMetricsSet`
+    ///
+    /// See [`ExpressionEvaluatorMetrics`] for details.
+    pub fn with_metrics(
+        &self,
+        metrics: &ExecutionPlanMetricsSet,
+        partition: usize,
+    ) -> Self {
+        let expr_metrics = self
+            .projection
+            .create_expression_metrics(metrics, partition);
+        Self {
+            expression_metrics: Some(expr_metrics),
+            projection: self.projection.clone(),
+            output_schema: Arc::clone(&self.output_schema),
+        }
+    }
+
     /// Project a record batch according to this projector's expressions.
     ///
     /// # Errors
     /// This function returns an error if any expression evaluation fails
     /// or if the output schema of the resulting record batch does not match
     /// the pre-computed output schema of the projector.
     pub fn project_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
-        let arrays = evaluate_expressions_to_arrays(
+        let arrays = evaluate_expressions_to_arrays_with_metrics(
             self.projection.exprs.iter().map(|p| &p.expr),
             batch,
+            self.expression_metrics.as_ref(),
         )?;
 
         if arrays.is_empty() {
diff --git a/datafusion/physical-plan/src/metrics.rs b/datafusion/physical-plan/src/metrics.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Metrics live in `datafusion-execution`; this module re-exports them to keep
-//! the public APIs stable.
+//! Metrics live in `datafusion-physical-expr-common`; this module re-exports
+//! them to keep the public APIs stable.
 
-pub use datafusion_execution::metrics::*;
+pub use datafusion_physical_expr_common::metrics::*;
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
@@ -295,8 +295,10 @@ impl ExecutionPlan for ProjectionExec {
             context.session_id(),
             context.task_id()
         );
+
+        let projector = self.projector.with_metrics(&self.metrics, partition);
         Ok(Box::pin(ProjectionStream::new(
-            self.projector.clone(),
+            projector,
             self.input.execute(partition, context)?,
             BaselineMetrics::new(&self.metrics, partition),
         )?))
@@ -1070,6 +1072,7 @@ mod tests {
     use std::sync::Arc;
 
     use crate::common::collect;
+
     use crate::test;
     use crate::test::exec::StatisticsExec;
 
diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt
@@ -23,5 +23,46 @@ EXPLAIN ANALYZE SELECT * FROM generate_series(100);
 ----
 Plan with Metrics LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=0, end=100, batch_size=8192], metrics=[output_rows=101, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
 
+# --------------------------------------------
+# Test ProjectionExec's per-expression metrics
+# --------------------------------------------
+
+statement ok
+set datafusion.explain.analyze_level = dev;
+
+# 1 expr
+# Expect metric `expr_0_eval_time` exists in ProjectionExec
+query TT
+EXPLAIN ANALYZE
+SELECT a
+FROM generate_series(1, 100) as t1(a);
+----
+Plan with Metrics
+01)ProjectionExec: expr=[value@0 as a], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=64.0 KB, output_batches=1, expr_0_eval_time=<slt:ignore>]
+<slt:ignore>
+
+# 2 exprs
+# Expect metrics `expr_0_eval_time` and `expr_1_eval_time` exist in ProjectionExec
+query TT
+EXPLAIN ANALYZE
+SELECT a+1, pow(a,2)
+FROM generate_series(1, 100) as t1(a);
+----
+Plan with Metrics
+01)ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=1632.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
+<slt:ignore>
+
+# common expressions
+# Expect metrics `expr_0_eval_time` and `expr_1_eval_time` exist in ProjectionExec
+query TT
+EXPLAIN ANALYZE
+SELECT a+1, a+1 as another_a_plus_one
+FROM generate_series(1, 100) as t1(a);
+----
+Plan with Metrics
+01)ProjectionExec: expr=[__common_expr_1@0 as t1.a + Int64(1), __common_expr_1@0 as another_a_plus_one], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
+02)--ProjectionExec: expr=[a@0 + 1 as __common_expr_1], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>]
+<slt:ignore>
+
 statement ok
 reset datafusion.explain.analyze_level;
