Add metric category filtering for EXPLAIN ANALYZE

Introduces `MetricCategory` (Rows, Bytes, Timing) so that EXPLAIN
ANALYZE output can be narrowed to only deterministic metrics, which is
especially useful in sqllogictest (.slt) files where timing values
would otherwise require `<slt:ignore>` markers everywhere.

Each `Metric` now optionally declares a category via
`MetricBuilder::with_category()`.  Well-known builder methods
(`output_rows`, `elapsed_compute`, …) set the category automatically;
custom counters/gauges default to "always included".

A new session config `datafusion.explain.analyze_categories` accepts
`all` (default), `none`, or a comma-separated list of `rows`, `bytes`,
`timing` to control which categories appear.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/common/src/config.rs

@@ -22,7 +22,7 @@ use arrow_ipc::CompressionType;
 #[cfg(feature = "parquet_encryption")]
 use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
 use crate::error::_config_err;
-use crate::format::{ExplainAnalyzeLevel, ExplainFormat};
+use crate::format::{ExplainAnalyzeCategories, ExplainAnalyzeLevel, ExplainFormat};
 use crate::parquet_config::DFParquetWriterVersion;
 use crate::parsers::CompressionTypeVariant;
 use crate::utils::get_available_parallelism;
@@ -1212,6 +1212,12 @@ config_namespace! {
         /// "summary" shows common metrics for high-level insights.
         /// "dev" provides deep operator-level introspection for developers.
         pub analyze_level: ExplainAnalyzeLevel, default = ExplainAnalyzeLevel::Dev
+
+        /// Which metric categories to include in "EXPLAIN ANALYZE" output.
+        /// Comma-separated list of: "rows", "bytes", "timing".
+        /// Use "none" to show plan structure only, or "all" (default) to show everything.
+        /// Metrics without a declared category are always shown (except with "none").
+        pub analyze_categories: ExplainAnalyzeCategories, default = ExplainAnalyzeCategories::All
     }
 }
 

datafusion/common/src/format.rs

@@ -206,6 +206,142 @@ impl ConfigField for ExplainFormat {
     }
 }
 
+/// Classifies a metric by what it measures.
+///
+/// This is orthogonal to [`MetricType`] (SUMMARY / DEV), which controls
+/// *verbosity*. `MetricCategory` controls *what kind of value* is shown,
+/// so that `EXPLAIN ANALYZE` output can be narrowed to only the categories
+/// that are useful in a given context.
+///
+/// For testing, the key property is **determinism**:
+/// - [`Rows`](Self::Rows) and [`Bytes`](Self::Bytes) depend on the plan
+///   and the data, so they are deterministic across runs (given the same
+///   input).
+/// - [`Timing`](Self::Timing) depends on hardware, system load, scheduling,
+///   etc., so it varies from run to run even on the same machine.
+///
+/// [`MetricCategory`] is especially useful in sqllogictest (`.slt`) files:
+/// setting `datafusion.explain.analyze_categories = 'rows'` lets a test
+/// assert on row-count metrics without sprinkling `<slt:ignore>` over every
+/// timing value.
+///
+/// Metrics that do not declare a category (the default for custom
+/// `Count` / `Gauge` metrics) are **always included** unless the config
+/// is set to `'none'`.
+///
+/// [`MetricType`]: datafusion_physical_expr_common::metrics::MetricType
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum MetricCategory {
+    /// Row counts and related dimensionless counters: `output_rows`,
+    /// `spilled_rows`, `output_batches`, pruning metrics, ratios, etc.
+    ///
+    /// Deterministic given the same plan and data.
+    Rows,
+    /// Byte measurements: `output_bytes`, `spilled_bytes`,
+    /// `current_memory_usage`, `bytes_scanned`, etc.
+    ///
+    /// Deterministic given the same plan and data.
+    Bytes,
+    /// Wall-clock durations and timestamps: `elapsed_compute`,
+    /// operator-defined `Time` metrics, `start_timestamp` /
+    /// `end_timestamp`, etc.
+    ///
+    /// **Non-deterministic** — varies across runs even on the same hardware.
+    Timing,
+}
+
+/// Controls which [`MetricCategory`] values are shown in `EXPLAIN ANALYZE`.
+///
+/// Set via `SET datafusion.explain.analyze_categories = '...'`.
+///
+/// See [`MetricCategory`] for the determinism properties that motivate
+/// this filter.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum ExplainAnalyzeCategories {
+    /// Show all metrics regardless of category (the default).
+    All,
+    /// Show only metrics whose category is in the list.
+    /// Metrics that have no declared category are still included
+    /// (they are treated as "always on").
+    ///
+    /// An **empty** vec means "plan only" — suppress all metrics.
+    Only(Vec<MetricCategory>),
+}
+
+impl Default for ExplainAnalyzeCategories {
+    fn default() -> Self {
+        Self::All
+    }
+}
+
+impl FromStr for ExplainAnalyzeCategories {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let s = s.trim().to_lowercase();
+        match s.as_str() {
+            "all" => Ok(Self::All),
+            "none" => Ok(Self::Only(vec![])),
+            other => {
+                let mut cats = Vec::new();
+                for part in other.split(',') {
+                    let part = part.trim();
+                    match part {
+                        "rows" => cats.push(MetricCategory::Rows),
+                        "bytes" => cats.push(MetricCategory::Bytes),
+                        "timing" => cats.push(MetricCategory::Timing),
+                        unknown => {
+                            return Err(DataFusionError::Configuration(format!(
+                                "Invalid metric category '{unknown}'. \
+                                 Expected 'all', 'none', or comma-separated list of \
+                                 'rows', 'bytes', 'timing'."
+                            )));
+                        }
+                    }
+                }
+                cats.dedup();
+                Ok(Self::Only(cats))
+            }
+        }
+    }
+}
+
+impl Display for ExplainAnalyzeCategories {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::All => write!(f, "all"),
+            Self::Only(cats) if cats.is_empty() => write!(f, "none"),
+            Self::Only(cats) => {
+                let mut first = true;
+                for cat in cats {
+                    if !first {
+                        write!(f, ",")?;
+                    }
+                    first = false;
+                    let s = match cat {
+                        MetricCategory::Rows => "rows",
+                        MetricCategory::Bytes => "bytes",
+                        MetricCategory::Timing => "timing",
+                    };
+                    write!(f, "{s}")?;
+                }
+                Ok(())
+            }
+        }
+    }
+}
+
+impl ConfigField for ExplainAnalyzeCategories {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = ExplainAnalyzeCategories::from_str(value)?;
+        Ok(())
+    }
+}
+
 /// Verbosity levels controlling how `EXPLAIN ANALYZE` renders metrics
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum ExplainAnalyzeLevel {

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -235,6 +235,7 @@ mod tests {
                 false,
                 false,
                 vec![MetricType::SUMMARY, MetricType::DEV],
+                None,
                 // use a new ParquetSource to avoid sharing execution metrics
                 self.build_parquet_exec(
                     file_group.clone(),

datafusion/core/src/physical_planner.rs

@@ -64,7 +64,7 @@ use arrow_schema::Field;
 use datafusion_catalog::ScanArgs;
 use datafusion_common::Column;
 use datafusion_common::display::ToStringifiedPlan;
-use datafusion_common::format::ExplainAnalyzeLevel;
+use datafusion_common::format::{ExplainAnalyzeCategories, ExplainAnalyzeLevel};
 use datafusion_common::tree_node::{
     Transformed, TreeNode, TreeNodeRecursion, TreeNodeVisitor,
 };
@@ -2720,10 +2720,17 @@ impl DefaultPhysicalPlanner {
             ExplainAnalyzeLevel::Summary => vec![MetricType::SUMMARY],
             ExplainAnalyzeLevel::Dev => vec![MetricType::SUMMARY, MetricType::DEV],
         };
+        let analyze_categories =
+            session_state.config_options().explain.analyze_categories.clone();
+        let metric_categories = match analyze_categories {
+            ExplainAnalyzeCategories::All => None,
+            ExplainAnalyzeCategories::Only(cats) => Some(cats),
+        };
         Ok(Arc::new(AnalyzeExec::new(
             a.verbose,
             show_statistics,
             metric_types,
+            metric_categories,
             input,
             schema,
         )))

datafusion/core/tests/sql/explain_analyze.rs

@@ -22,7 +22,9 @@ use rstest::rstest;
 use datafusion::config::ConfigOptions;
 use datafusion::physical_plan::display::DisplayableExecutionPlan;
 use datafusion::physical_plan::metrics::Timestamp;
-use datafusion_common::format::ExplainAnalyzeLevel;
+use datafusion_common::format::{
+    ExplainAnalyzeCategories, ExplainAnalyzeLevel, MetricCategory,
+};
 use object_store::path::Path;
 
 #[tokio::test]
@@ -219,6 +221,23 @@ async fn collect_plan_with_context(
         .to_string()
 }
 
+async fn collect_plan_with_categories(
+    sql_str: &str,
+    categories: ExplainAnalyzeCategories,
+) -> String {
+    let ctx = SessionContext::new();
+    {
+        let state = ctx.state_ref();
+        let mut state = state.write();
+        state.config_mut().options_mut().explain.analyze_categories = categories;
+    }
+    let dataframe = ctx.sql(sql_str).await.unwrap();
+    let batches = dataframe.collect().await.unwrap();
+    arrow::util::pretty::pretty_format_batches(&batches)
+        .unwrap()
+        .to_string()
+}
+
 async fn collect_plan(sql_str: &str, level: ExplainAnalyzeLevel) -> String {
     let ctx = SessionContext::new();
     collect_plan_with_context(sql_str, &ctx, level).await
@@ -1160,3 +1179,78 @@ async fn explain_analyze_hash_join() {
         );
     }
 }
+
+#[tokio::test]
+async fn explain_analyze_categories() {
+    let sql = "EXPLAIN ANALYZE \
+            SELECT * \
+            FROM generate_series(10) as t1(v1) \
+            ORDER BY v1 DESC";
+
+    for (categories, needle, should_contain) in [
+        // "rows" category: output_rows yes, elapsed_compute no, output_bytes no
+        (
+            ExplainAnalyzeCategories::Only(vec![MetricCategory::Rows]),
+            "output_rows",
+            true,
+        ),
+        (
+            ExplainAnalyzeCategories::Only(vec![MetricCategory::Rows]),
+            "elapsed_compute",
+            false,
+        ),
+        (
+            ExplainAnalyzeCategories::Only(vec![MetricCategory::Rows]),
+            "output_bytes",
+            false,
+        ),
+        // "none" — plan only, no metrics at all
+        (
+            ExplainAnalyzeCategories::Only(vec![]),
+            "output_rows",
+            false,
+        ),
+        (
+            ExplainAnalyzeCategories::Only(vec![]),
+            "elapsed_compute",
+            false,
+        ),
+        // "all" — everything shown
+        (ExplainAnalyzeCategories::All, "output_rows", true),
+        (ExplainAnalyzeCategories::All, "elapsed_compute", true),
+        (ExplainAnalyzeCategories::All, "output_bytes", true),
+        // "rows,bytes" — row + byte metrics, no timing
+        (
+            ExplainAnalyzeCategories::Only(vec![
+                MetricCategory::Rows,
+                MetricCategory::Bytes,
+            ]),
+            "output_rows",
+            true,
+        ),
+        (
+            ExplainAnalyzeCategories::Only(vec![
+                MetricCategory::Rows,
+                MetricCategory::Bytes,
+            ]),
+            "output_bytes",
+            true,
+        ),
+        (
+            ExplainAnalyzeCategories::Only(vec![
+                MetricCategory::Rows,
+                MetricCategory::Bytes,
+            ]),
+            "elapsed_compute",
+            false,
+        ),
+    ] {
+        let plan = collect_plan_with_categories(sql, categories.clone()).await;
+        assert_eq!(
+            plan.contains(needle),
+            should_contain,
+            "plan for categories {categories:?} should{} contain '{needle}':\n{plan}",
+            if should_contain { "" } else { " NOT" }
+        );
+    }
+}