diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index 7d32f2a88fd9c..1f1159e0c919f 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -1385,6 +1385,7 @@ config_namespace! {
         /// Verbosity level for "EXPLAIN ANALYZE". Default is "dev"
         /// "summary" shows common metrics for high-level insights.
         /// "dev" provides deep operator-level introspection for developers.
+        /// "internal" provides low-level kernel debugging metrics.
         pub analyze_level: MetricType, default = MetricType::Dev
 
         /// Which metric categories to include in "EXPLAIN ANALYZE" output.
diff --git a/datafusion/common/src/format.rs b/datafusion/common/src/format.rs
index a6bd42be691a9..2d5fc5c5f5b0d 100644
--- a/datafusion/common/src/format.rs
+++ b/datafusion/common/src/format.rs
@@ -210,7 +210,8 @@ impl ConfigField for ExplainFormat {
 ///
 /// The `datafusion.explain.analyze_level` configuration controls which
 /// type is shown:
-/// - `"dev"` (the default): all metrics are shown.
+/// - `"dev"` (the default): summary and developer-facing metrics are shown.
+/// - `"internal"`: all metrics are shown, including kernel debugging metrics.
 /// - `"summary"`: only metrics tagged as `Summary` are shown.
 ///
 /// This is orthogonal to [`MetricCategory`], which filters by *what kind*
@@ -228,19 +229,30 @@ pub enum MetricType {
     Summary,
     /// For deep operator-level introspection for developers
     Dev,
+    /// For low-level kernel debugging and DataFusion development
+    Internal,
 }
 
 impl MetricType {
     /// Returns the set of metric types that should be shown for this level.
     ///
-    /// `Dev` is a superset of `Summary`: when the user selects
-    /// `analyze_level = 'dev'`, both `Summary` and `Dev` metrics are shown.
+    /// `Dev` is a superset of `Summary`, and `Internal` is a superset of
+    /// both: when the user selects `analyze_level = 'internal'`, `Summary`,
+    /// `Dev`, and `Internal` metrics are shown.
     pub fn included_types(self) -> Vec<MetricType> {
         match self {
             MetricType::Summary => vec![MetricType::Summary],
             MetricType::Dev => vec![MetricType::Summary, MetricType::Dev],
+            MetricType::Internal => {
+                vec![MetricType::Summary, MetricType::Dev, MetricType::Internal]
+            }
         }
     }
+
+    /// Returns true if `self` includes metrics tagged as `metric_type`.
+    pub fn includes(self, metric_type: MetricType) -> bool {
+        self.included_types().contains(&metric_type)
+    }
 }
 
 impl FromStr for MetricType {
@@ -250,8 +262,9 @@ impl FromStr for MetricType {
         match s.trim().to_lowercase().as_str() {
             "summary" => Ok(Self::Summary),
             "dev" => Ok(Self::Dev),
+            "internal" => Ok(Self::Internal),
             other => Err(DataFusionError::Configuration(format!(
-                "Invalid explain analyze level. Expected 'summary' or 'dev'. Got '{other}'"
+                "Invalid explain analyze level. Expected 'summary', 'dev', or 'internal'. Got '{other}'"
             ))),
         }
     }
@@ -262,6 +275,7 @@ impl Display for MetricType {
         match self {
             Self::Summary => write!(f, "summary"),
             Self::Dev => write!(f, "dev"),
+            Self::Internal => write!(f, "internal"),
         }
     }
 }
diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs
index b093563d9adda..a2a0d1847eafe 100644
--- a/datafusion/core/tests/sql/explain_analyze.rs
+++ b/datafusion/core/tests/sql/explain_analyze.rs
@@ -236,6 +236,25 @@ async fn collect_plan_with_categories(
         .to_string()
 }
 
+async fn collect_plan_after_set(
+    ctx: &SessionContext,
+    level: &str,
+    sql_str: &str,
+) -> String {
+    ctx.sql(&format!("SET datafusion.explain.analyze_level = '{level}'"))
+        .await
+        .unwrap()
+        .collect()
+        .await
+        .unwrap();
+
+    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: MetricType) -> String {
     let ctx = SessionContext::new();
     collect_plan_with_context(sql_str, &ctx, level).await
@@ -257,6 +276,10 @@ async fn explain_analyze_level() {
         (MetricType::Dev, "output_rows", true),
         (MetricType::Dev, "output_bytes", true),
         (MetricType::Dev, "output_batches", true),
+        (MetricType::Internal, "spill_count", true),
+        (MetricType::Internal, "output_rows", true),
+        (MetricType::Internal, "output_bytes", true),
+        (MetricType::Internal, "output_batches", true),
     ] {
         let plan = collect_plan(sql, level).await;
         assert_eq!(
@@ -267,6 +290,33 @@ async fn explain_analyze_level() {
     }
 }
 
+#[tokio::test]
+async fn explain_analyze_internal_repartition_metrics() {
+    let ctx = SessionContext::new_with_config(
+        SessionConfig::new()
+            .with_target_partitions(4)
+            .with_batch_size(4096),
+    );
+    register_aggregate_csv_by_sql(&ctx).await;
+
+    let sql = "EXPLAIN ANALYZE \
+        SELECT c1, count(*) \
+        FROM aggregate_test_100 \
+        GROUP BY c1";
+
+    let dev_plan = collect_plan_after_set(&ctx, "dev", sql).await;
+    assert_contains!(&dev_plan, "RepartitionExec");
+    assert_not_contains!(&dev_plan, "hash_compute_time");
+    assert_not_contains!(&dev_plan, "route_time");
+    assert_not_contains!(&dev_plan, "batch_build_time");
+
+    let internal_plan = collect_plan_after_set(&ctx, "internal", sql).await;
+    assert_contains!(&internal_plan, "RepartitionExec");
+    assert_contains!(&internal_plan, "hash_compute_time");
+    assert_contains!(&internal_plan, "route_time");
+    assert_contains!(&internal_plan, "batch_build_time");
+}
+
 #[tokio::test]
 async fn explain_analyze_level_datasource_parquet() {
     let table_name = "tpch_lineitem_small";
@@ -284,6 +334,8 @@ async fn explain_analyze_level_datasource_parquet() {
         (MetricType::Summary, "page_index_eval_time", false),
         (MetricType::Dev, "metadata_load_time", true),
         (MetricType::Dev, "page_index_eval_time", true),
+        (MetricType::Internal, "metadata_load_time", true),
+        (MetricType::Internal, "page_index_eval_time", true),
     ] {
         let plan = collect_plan_with_context(&sql, &ctx, level).await;
 
diff --git a/datafusion/physical-expr-common/src/metrics/builder.rs b/datafusion/physical-expr-common/src/metrics/builder.rs
index e9c0b76af2582..42758dc438491 100644
--- a/datafusion/physical-expr-common/src/metrics/builder.rs
+++ b/datafusion/physical-expr-common/src/metrics/builder.rs
@@ -117,6 +117,19 @@ impl<'a> MetricBuilder<'a> {
         self
     }
 
+    /// Switches to a conditional builder for metrics controlled by
+    /// `datafusion.explain.analyze_level`.
+    ///
+    /// The returned builder preserves this builder's labels, category, and
+    /// [`MetricType`], but its terminal methods return `Option<T>` and skip
+    /// registration when `enabled_level` does not include this builder's
+    /// [`MetricType`].
+    pub fn if_enabled(self, enabled_level: MetricType) -> ConditionalMetricBuilder<'a> {
+        ConditionalMetricBuilder {
+            builder: enabled_level.includes(self.metric_type).then_some(self),
+        }
+    }
+
     /// Consume self and create a metric of the specified value
     /// registered with the MetricsSet
     pub fn build(self, value: MetricValue) {
@@ -333,3 +346,51 @@ impl<'a> MetricBuilder<'a> {
         ratio_metrics
     }
 }
+
+/// A conditional metric builder used after [`MetricBuilder::if_enabled`].
+///
+/// This type intentionally mirrors a subset of [`MetricBuilder`], but its
+/// terminal methods return `Option<T>`.
+/// - `Some(T)` means the metric was registered and can be updated.
+/// - `None` means the configured metric level does not include this metric,
+///   so the caller should do no metric work.
+pub struct ConditionalMetricBuilder<'a> {
+    builder: Option<MetricBuilder<'a>>,
+}
+
+impl<'a> ConditionalMetricBuilder<'a> {
+    /// Add a label to the metric being constructed, if this builder is enabled.
+    pub fn with_label(mut self, label: Label) -> Self {
+        self.builder = self.builder.map(|builder| builder.with_label(label));
+        self
+    }
+
+    /// Set the semantic category for the metric being constructed, if this
+    /// builder is enabled.
+    pub fn with_category(mut self, category: MetricCategory) -> Self {
+        self.builder = self.builder.map(|builder| builder.with_category(category));
+        self
+    }
+
+    /// Consume self and conditionally create a new timer for recording some
+    /// subset of an operator's execution time.
+    pub fn subset_time(
+        self,
+        subset_name: impl Into<Cow<'static, str>>,
+        partition: usize,
+    ) -> Option<Time> {
+        self.builder
+            .map(|builder| builder.subset_time(subset_name, partition))
+    }
+
+    /// Consume self and conditionally create a new counter for recording some
+    /// arbitrary metric of an operator.
+    pub fn counter(
+        self,
+        counter_name: impl Into<Cow<'static, str>>,
+        partition: usize,
+    ) -> Option<Count> {
+        self.builder
+            .map(|builder| builder.counter(counter_name, partition))
+    }
+}
diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs
index eecd8cfabd5eb..f10dd12441075 100644
--- a/datafusion/physical-expr-common/src/metrics/mod.rs
+++ b/datafusion/physical-expr-common/src/metrics/mod.rs
@@ -623,6 +623,61 @@ mod tests {
         assert_eq!(metrics.clone_inner().output_rows().unwrap(), 20);
     }
 
+    #[test]
+    fn test_conditional_metric_builder_uses_metric_type_level() {
+        for (enabled_level, metric_type, should_register) in [
+            (MetricType::Summary, MetricType::Summary, true),
+            (MetricType::Summary, MetricType::Dev, false),
+            (MetricType::Summary, MetricType::Internal, false),
+            (MetricType::Dev, MetricType::Summary, true),
+            (MetricType::Dev, MetricType::Dev, true),
+            (MetricType::Dev, MetricType::Internal, false),
+            (MetricType::Internal, MetricType::Summary, true),
+            (MetricType::Internal, MetricType::Dev, true),
+            (MetricType::Internal, MetricType::Internal, true),
+        ] {
+            let metrics = ExecutionPlanMetricsSet::new();
+            let metric = MetricBuilder::new(&metrics)
+                .with_type(metric_type)
+                .if_enabled(enabled_level)
+                .subset_time("conditional_time", 0);
+
+            assert_eq!(
+                metric.is_some(),
+                should_register,
+                "enabled_level={enabled_level:?} metric_type={metric_type:?}"
+            );
+            assert_eq!(
+                metrics
+                    .clone_inner()
+                    .sum_by_name("conditional_time")
+                    .is_some(),
+                should_register,
+                "enabled_level={enabled_level:?} metric_type={metric_type:?}"
+            );
+        }
+
+        let metrics = ExecutionPlanMetricsSet::new();
+        let label = Label::new("label", "value");
+        let metric = MetricBuilder::new(&metrics)
+            .with_type(MetricType::Internal)
+            .with_category(MetricCategory::Timing)
+            .if_enabled(MetricType::Internal)
+            .with_label(label.clone())
+            .subset_time("internal_time", 0);
+
+        assert!(metric.is_some());
+
+        let metrics = metrics.clone_inner();
+        let metric = metrics
+            .iter()
+            .find(|metric| metric.value().name() == "internal_time")
+            .expect("registered internal metric");
+        assert_eq!(metric.metric_type(), MetricType::Internal);
+        assert_eq!(metric.metric_category(), Some(MetricCategory::Timing));
+        assert_eq!(metric.labels(), &[label]);
+    }
+
     #[test]
     fn test_elapsed_compute() {
         let metrics = ExecutionPlanMetricsSet::new();
diff --git a/datafusion/physical-plan/src/analyze.rs b/datafusion/physical-plan/src/analyze.rs
index ea3abf439e4c1..5824464baa79d 100644
--- a/datafusion/physical-plan/src/analyze.rs
+++ b/datafusion/physical-plan/src/analyze.rs
@@ -94,6 +94,11 @@ impl AnalyzeExec {
         self.metric_categories.as_deref()
     }
 
+    /// Access to metric_types
+    pub fn metric_types(&self) -> &[MetricType] {
+        &self.metric_types
+    }
+
     /// The input plan
     pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
         &self.input
@@ -187,6 +192,8 @@ impl ExecutionPlan for AnalyzeExec {
         let num_input_partitions = self.input.output_partitioning().partition_count();
         let mut builder =
             RecordBatchReceiverStream::builder(self.schema(), num_input_partitions);
+        let metric_types = self.metric_types.clone();
+        let context = task_context_with_metric_level(context, &metric_types);
 
         for input_partition in 0..num_input_partitions {
             builder.run_input(
@@ -202,7 +209,6 @@ impl ExecutionPlan for AnalyzeExec {
         let captured_schema = Arc::clone(&self.schema);
         let verbose = self.verbose;
         let show_statistics = self.show_statistics;
-        let metric_types = self.metric_types.clone();
         let metric_categories = self.metric_categories.clone();
 
         // future that gathers the results from all the tasks in the
@@ -292,11 +298,48 @@ fn create_output_batch(
     .map_err(DataFusionError::from)
 }
 
+fn task_context_with_metric_level(
+    context: Arc<TaskContext>,
+    metric_types: &[MetricType],
+) -> Arc<TaskContext> {
+    let analyze_level = analyze_level_from_metric_types(metric_types);
+    if context.session_config().options().explain.analyze_level == analyze_level {
+        return context;
+    }
+
+    let mut session_config = context.session_config().clone();
+    session_config.options_mut().explain.analyze_level = analyze_level;
+
+    Arc::new(TaskContext::new(
+        context.task_id(),
+        context.session_id(),
+        session_config,
+        context.scalar_functions().clone(),
+        context.higher_order_functions().clone(),
+        context.aggregate_functions().clone(),
+        context.window_functions().clone(),
+        context.runtime_env(),
+    ))
+}
+
+fn analyze_level_from_metric_types(metric_types: &[MetricType]) -> MetricType {
+    if metric_types.contains(&MetricType::Internal) {
+        MetricType::Internal
+    } else if metric_types.contains(&MetricType::Dev) {
+        MetricType::Dev
+    } else {
+        MetricType::Summary
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::{
         collect,
+        empty::EmptyExec,
+        expressions::col,
+        repartition::RepartitionExec,
         test::{
             assert_is_pending,
             exec::{BlockingExec, assert_strong_count_converges_to_zero},
@@ -304,6 +347,7 @@ mod tests {
     };
 
     use arrow::datatypes::{DataType, Field, Schema};
+    use arrow::util::pretty::pretty_format_batches;
     use futures::FutureExt;
 
     #[tokio::test]
@@ -332,4 +376,39 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn analyze_exec_metric_types_drive_child_metric_registration() -> Result<()> {
+        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
+        let input = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+        let repartition = Arc::new(RepartitionExec::try_new(
+            input,
+            Partitioning::Hash(vec![col("i", &schema)?], 2),
+        )?);
+
+        let analyze_schema = Arc::new(Schema::new(vec![
+            Field::new("plan_type", DataType::Utf8, false),
+            Field::new("plan", DataType::Utf8, false),
+        ]));
+        let analyze_exec = Arc::new(AnalyzeExec::new(
+            false,
+            false,
+            vec![MetricType::Summary, MetricType::Dev, MetricType::Internal],
+            None,
+            repartition,
+            analyze_schema,
+        ));
+
+        // TaskContext defaults to analyze_level = dev. The AnalyzeExec setting
+        // must still drive internal metric registration in the analyzed child.
+        let batches = collect(analyze_exec, Arc::new(TaskContext::default())).await?;
+        let formatted = pretty_format_batches(&batches)?.to_string();
+
+        assert!(
+            formatted.contains("hash_compute_time"),
+            "internal repartition metrics should be registered from AnalyzeExec metric_types: {formatted}"
+        );
+
+        Ok(())
+    }
 }
diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs
index 873ba8a5ee487..0adb5c2684a9c 100644
--- a/datafusion/physical-plan/src/repartition/mod.rs
+++ b/datafusion/physical-plan/src/repartition/mod.rs
@@ -239,7 +239,7 @@ impl RepartitionExecState {
         &mut self,
         input: &Arc<dyn ExecutionPlan>,
         metrics: &ExecutionPlanMetricsSet,
-        output_partitions: usize,
+        partitioning: &Partitioning,
         ctx: &Arc<TaskContext>,
     ) -> Result<()> {
         if !matches!(self, RepartitionExecState::NotInitialized) {
@@ -247,10 +247,12 @@ impl RepartitionExecState {
         }
 
         let num_input_partitions = input.output_partitioning().partition_count();
+        let analyze_level = ctx.session_config().options().explain.analyze_level;
         let mut streams_and_metrics = Vec::with_capacity(num_input_partitions);
 
         for i in 0..num_input_partitions {
-            let metrics = RepartitionMetrics::new(i, output_partitions, metrics);
+            let metrics =
+                RepartitionMetrics::new(i, partitioning, metrics, analyze_level);
 
             let timer = metrics.fetch_time.timer();
             let stream = input.execute(i, Arc::clone(ctx))?;
@@ -278,7 +280,7 @@ impl RepartitionExecState {
                 self.ensure_input_streams_initialized(
                     input,
                     metrics,
-                    partitioning.partition_count(),
+                    partitioning,
                     context,
                 )?;
                 let RepartitionExecState::InputStreamsInitialized(value) = self else {
@@ -418,7 +420,7 @@ impl RepartitionExecState {
 /// A utility that can be used to partition batches based on [`Partitioning`]
 pub struct BatchPartitioner {
     state: BatchPartitionerState,
-    timer: metrics::Time,
+    metrics: BatchPartitionerMetrics,
 }
 
 enum BatchPartitionerState {
@@ -434,6 +436,29 @@ enum BatchPartitionerState {
     },
 }
 
+#[derive(Debug, Clone)]
+struct BatchPartitionerMetrics {
+    /// End-to-end time spent repartitioning a batch.
+    total_time: metrics::Time,
+    /// Time spent evaluating partitioning expressions and computing hashes.
+    hash_compute_time: Option<metrics::Time>,
+    /// Time spent routing row indices to output partitions.
+    route_time: Option<metrics::Time>,
+    /// Time spent materializing partitioned output batches from routed indices.
+    batch_build_time: Option<metrics::Time>,
+}
+
+impl BatchPartitionerMetrics {
+    fn new(total_time: metrics::Time) -> Self {
+        Self {
+            total_time,
+            hash_compute_time: None,
+            route_time: None,
+            batch_build_time: None,
+        }
+    }
+}
+
 /// Fixed RandomState used for hash repartitioning to ensure consistent behavior across
 /// executions and runs.
 pub const REPARTITION_RANDOM_STATE: SeededRandomState = SeededRandomState::with_seed(0);
@@ -523,10 +548,10 @@ impl BatchPartitioner {
     ///
     /// # Errors
     /// Returns an error if `num_partitions` is zero.
-    pub fn new_hash_partitioner(
+    fn new_hash_partitioner_with_metrics(
         exprs: Vec<Arc<dyn PhysicalExpr>>,
         num_partitions: usize,
-        timer: metrics::Time,
+        metrics: BatchPartitionerMetrics,
     ) -> Result<Self> {
         if num_partitions == 0 {
             return internal_err!("Hash repartition requires at least one partition");
@@ -539,10 +564,34 @@ impl BatchPartitioner {
                 hash_buffer: vec![],
                 indices: vec![vec![]; num_partitions],
             },
-            timer,
+            metrics,
         })
     }
 
+    /// Create a new [`BatchPartitioner`] for hash-based repartitioning.
+    ///
+    /// # Parameters
+    /// - `exprs`: Expressions used to compute the hash for each input row.
+    /// - `num_partitions`: Total number of output partitions.
+    /// - `timer`: Metric used to record time spent during repartitioning.
+    ///
+    /// The partition count is fixed for the lifetime of the partitioner, so this
+    /// precomputes a strength-reduced reducer for `hash % num_partitions`.
+    ///
+    /// # Errors
+    /// Returns an error if `num_partitions` is zero.
+    pub fn new_hash_partitioner(
+        exprs: Vec<Arc<dyn PhysicalExpr>>,
+        num_partitions: usize,
+        timer: metrics::Time,
+    ) -> Result<Self> {
+        Self::new_hash_partitioner_with_metrics(
+            exprs,
+            num_partitions,
+            BatchPartitionerMetrics::new(timer),
+        )
+    }
+
     /// Create a new [`BatchPartitioner`] for round-robin repartitioning.
     ///
     /// # Parameters
@@ -554,9 +603,9 @@ impl BatchPartitioner {
     /// # Notes
     /// The starting output partition is derived from the input partition
     /// to avoid skew when multiple input partitions are used.
-    pub fn new_round_robin_partitioner(
+    fn new_round_robin_partitioner_with_metrics(
         num_partitions: usize,
-        timer: metrics::Time,
+        metrics: BatchPartitionerMetrics,
         input_partition: usize,
         num_input_partitions: usize,
     ) -> Self {
@@ -565,9 +614,34 @@ impl BatchPartitioner {
                 num_partitions,
                 next_idx: (input_partition * num_partitions) / num_input_partitions,
             },
-            timer,
+            metrics,
         }
     }
+
+    /// Create a new [`BatchPartitioner`] for round-robin repartitioning.
+    ///
+    /// # Parameters
+    /// - `num_partitions`: Total number of output partitions.
+    /// - `timer`: Metric used to record time spent during repartitioning.
+    /// - `input_partition`: Index of the current input partition.
+    /// - `num_input_partitions`: Total number of input partitions.
+    ///
+    /// # Notes
+    /// The starting output partition is derived from the input partition
+    /// to avoid skew when multiple input partitions are used.
+    pub fn new_round_robin_partitioner(
+        num_partitions: usize,
+        timer: metrics::Time,
+        input_partition: usize,
+        num_input_partitions: usize,
+    ) -> Self {
+        Self::new_round_robin_partitioner_with_metrics(
+            num_partitions,
+            BatchPartitionerMetrics::new(timer),
+            input_partition,
+            num_input_partitions,
+        )
+    }
     /// Create a new [`BatchPartitioner`] based on the provided [`Partitioning`] scheme.
     ///
     /// This is a convenience constructor that delegates to the specialized
@@ -649,6 +723,7 @@ impl BatchPartitioner {
                     num_partitions,
                     next_idx,
                 } => {
+                    let _total_time_scope = self.metrics.total_time.timer();
                     let idx = *next_idx;
                     *next_idx = (*next_idx + 1) % *num_partitions;
                     Box::new(std::iter::once(Ok((idx, batch))))
@@ -659,30 +734,34 @@ impl BatchPartitioner {
                     hash_buffer,
                     indices,
                 } => {
-                    // Tracking time required for distributing indexes across output partitions
-                    let timer = self.timer.timer();
-
-                    let arrays =
-                        evaluate_expressions_to_arrays(exprs.as_slice(), &batch)?;
-
-                    hash_buffer.clear();
-                    hash_buffer.resize(batch.num_rows(), 0);
-
-                    create_hashes(
-                        &arrays,
-                        REPARTITION_RANDOM_STATE.random_state(),
-                        hash_buffer,
-                    )?;
-
-                    indices.iter_mut().for_each(|v| v.clear());
-
-                    partition_reducer.partition_indices(hash_buffer, indices);
-
-                    // Finished building index-arrays for output partitions
-                    timer.done();
+                    let _total_time_scope = self.metrics.total_time.timer();
+
+                    {
+                        let _hash_compute_time_scope = self
+                            .metrics
+                            .hash_compute_time
+                            .as_ref()
+                            .map(|time| time.timer());
+
+                        let arrays =
+                            evaluate_expressions_to_arrays(exprs.as_slice(), &batch)?;
+
+                        hash_buffer.clear();
+                        hash_buffer.resize(batch.num_rows(), 0);
+
+                        create_hashes(
+                            &arrays,
+                            REPARTITION_RANDOM_STATE.random_state(),
+                            hash_buffer,
+                        )?;
+                    }
 
-                    // Borrowing partitioner timer to prevent moving `self` to closure
-                    let partitioner_timer = &self.timer;
+                    {
+                        let _route_time_scope =
+                            self.metrics.route_time.as_ref().map(|time| time.timer());
+                        indices.iter_mut().for_each(|v| v.clear());
+                        partition_reducer.partition_indices(hash_buffer, indices);
+                    }
 
                     let mut partitioned_batches = vec![];
                     for (partition, p_indices) in indices.iter_mut().enumerate() {
@@ -691,12 +770,14 @@ impl BatchPartitioner {
                             let indices_array: PrimitiveArray<UInt32Type> =
                                 taken_indices.into();
 
-                            // Tracking time required for repartitioned batches construction
-                            let _timer = partitioner_timer.timer();
-
-                            // Produce batches based on indices
-                            let columns =
-                                take_arrays(batch.columns(), &indices_array, None)?;
+                            let columns = {
+                                let _batch_build_time_scope = self
+                                    .metrics
+                                    .batch_build_time
+                                    .as_ref()
+                                    .map(|time| time.timer());
+                                take_arrays(batch.columns(), &indices_array, None)?
+                            };
 
                             let mut options = RecordBatchOptions::new();
                             options = options.with_row_count(Some(indices_array.len()));
@@ -862,18 +943,32 @@ struct RepartitionMetrics {
     fetch_time: metrics::Time,
     /// Repartitioning elapsed time in nanos
     repartition_time: metrics::Time,
+    /// Time in nanos spent evaluating partitioning expressions and computing hashes.
+    hash_compute_time: Option<metrics::Time>,
+    /// Time in nanos spent routing row indices to output partitions.
+    route_time: Option<metrics::Time>,
+    /// Time in nanos spent constructing output batches.
+    batch_build_time: Option<metrics::Time>,
     /// Time in nanos for sending resulting batches to channels.
     ///
     /// One metric per output partition.
     send_time: Vec<metrics::Time>,
+    /// Time in nanos spent waiting for channel capacity.
+    channel_wait_time: Vec<metrics::Time>,
+    /// Time in nanos spent writing spilled batches.
+    spill_write_time: Vec<metrics::Time>,
 }
 
 impl RepartitionMetrics {
     pub fn new(
         input_partition: usize,
-        num_output_partitions: usize,
+        partitioning: &Partitioning,
         metrics: &ExecutionPlanMetricsSet,
+        analyze_level: metrics::MetricType,
     ) -> Self {
+        let num_output_partitions = partitioning.partition_count();
+        let register_hash_metrics = matches!(partitioning, Partitioning::Hash(_, _));
+
         // Time in nanos to execute child operator and fetch batches
         let fetch_time =
             MetricBuilder::new(metrics).subset_time("fetch_time", input_partition);
@@ -882,6 +977,36 @@ impl RepartitionMetrics {
         let repartition_time =
             MetricBuilder::new(metrics).subset_time("repartition_time", input_partition);
 
+        // Time in nanos to evaluate partition expressions and compute row hashes
+        let hash_compute_time = register_hash_metrics
+            .then(|| {
+                MetricBuilder::new(metrics)
+                    .with_type(metrics::MetricType::Internal)
+                    .if_enabled(analyze_level)
+                    .subset_time("hash_compute_time", input_partition)
+            })
+            .flatten();
+
+        // Time in nanos to route row indices to output partitions
+        let route_time = register_hash_metrics
+            .then(|| {
+                MetricBuilder::new(metrics)
+                    .with_type(metrics::MetricType::Internal)
+                    .if_enabled(analyze_level)
+                    .subset_time("route_time", input_partition)
+            })
+            .flatten();
+
+        // Time in nanos to build output batches from routed row indices
+        let batch_build_time = register_hash_metrics
+            .then(|| {
+                MetricBuilder::new(metrics)
+                    .with_type(metrics::MetricType::Internal)
+                    .if_enabled(analyze_level)
+                    .subset_time("batch_build_time", input_partition)
+            })
+            .flatten();
+
         // Time in nanos for sending resulting batches to channels
         let send_time = (0..num_output_partitions)
             .map(|output_partition| {
@@ -893,10 +1018,50 @@ impl RepartitionMetrics {
             })
             .collect();
 
+        // Time in nanos waiting for output channel capacity
+        let channel_wait_time = (0..num_output_partitions)
+            .filter_map(|output_partition| {
+                let label =
+                    metrics::Label::new("outputPartition", output_partition.to_string());
+                MetricBuilder::new(metrics)
+                    .with_type(metrics::MetricType::Internal)
+                    .if_enabled(analyze_level)
+                    .with_label(label)
+                    .subset_time("channel_wait_time", input_partition)
+            })
+            .collect();
+
+        // Time in nanos writing spilled batches to spill storage
+        let spill_write_time = (0..num_output_partitions)
+            .filter_map(|output_partition| {
+                let label =
+                    metrics::Label::new("outputPartition", output_partition.to_string());
+                MetricBuilder::new(metrics)
+                    .with_type(metrics::MetricType::Internal)
+                    .if_enabled(analyze_level)
+                    .with_label(label)
+                    .subset_time("spill_write_time", input_partition)
+            })
+            .collect();
+
         Self {
             fetch_time,
             repartition_time,
+            hash_compute_time,
+            route_time,
+            batch_build_time,
             send_time,
+            channel_wait_time,
+            spill_write_time,
+        }
+    }
+
+    fn batch_partitioner_metrics(&self) -> BatchPartitionerMetrics {
+        BatchPartitionerMetrics {
+            total_time: self.repartition_time.clone(),
+            hash_compute_time: self.hash_compute_time.clone(),
+            route_time: self.route_time.clone(),
+            batch_build_time: self.batch_build_time.clone(),
         }
     }
 }
@@ -1070,7 +1235,7 @@ impl ExecutionPlan for RepartitionExec {
             state.ensure_input_streams_initialized(
                 &input,
                 &metrics,
-                partitioning.partition_count(),
+                &partitioning,
                 &context,
             )?;
         }
@@ -1431,18 +1596,19 @@ impl RepartitionExec {
         input_partition: usize,
         num_input_partitions: usize,
     ) -> Result<()> {
+        let batch_partitioner_metrics = metrics.batch_partitioner_metrics();
         let mut partitioner = match &partitioning {
             Partitioning::Hash(exprs, num_partitions) => {
-                BatchPartitioner::new_hash_partitioner(
+                BatchPartitioner::new_hash_partitioner_with_metrics(
                     exprs.clone(),
                     *num_partitions,
-                    metrics.repartition_time.clone(),
+                    batch_partitioner_metrics.clone(),
                 )?
             }
             Partitioning::RoundRobinBatch(num_partitions) => {
-                BatchPartitioner::new_round_robin_partitioner(
+                BatchPartitioner::new_round_robin_partitioner_with_metrics(
                     *num_partitions,
-                    metrics.repartition_time.clone(),
+                    batch_partitioner_metrics,
                     input_partition,
                     num_input_partitions,
                 )
@@ -1487,13 +1653,25 @@ impl RepartitionExec {
                             Err(_) => {
                                 // We're memory limited - spill to SpillPool
                                 // SpillPool handles file handle reuse and rotation
+                                let _spill_write_time_scope = metrics
+                                    .spill_write_time
+                                    .get(partition)
+                                    .map(|time| time.timer());
                                 channel.spill_writer.push_batch(&batch)?;
                                 // Send marker indicating batch was spilled
                                 (RepartitionBatch::Spilled, false)
                             }
                         };
 
-                    if channel.sender.send(Some(Ok(batch_to_send))).await.is_err() {
+                    let send_err = {
+                        let _channel_wait_time_scope = metrics
+                            .channel_wait_time
+                            .get(partition)
+                            .map(|time| time.timer());
+                        channel.sender.send(Some(Ok(batch_to_send))).await.is_err()
+                    };
+
+                    if send_err {
                         // If the other end has hung up, it was an early shutdown (e.g. LIMIT)
                         // Only shrink memory if it was a memory batch
                         if is_memory_batch {
@@ -2043,6 +2221,121 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn repartition_internal_metrics_require_internal_level() -> Result<()> {
+        let schema = test_schema();
+        for level in [metrics::MetricType::Summary, metrics::MetricType::Dev] {
+            let metrics = repartition_metric_names(
+                level,
+                Partitioning::Hash(vec![col("c0", &schema)?], 2),
+            )
+            .await?;
+            assert!(
+                !metrics.contains("hash_compute_time"),
+                "{level:?} metrics should not include repartition internals: {metrics:?}"
+            );
+            assert!(
+                !metrics.contains("route_time"),
+                "{level:?} metrics should not include repartition internals: {metrics:?}"
+            );
+            assert!(
+                !metrics.contains("channel_wait_time"),
+                "{level:?} metrics should not include repartition internals: {metrics:?}"
+            );
+            assert!(
+                !metrics.contains("spill_write_time"),
+                "{level:?} metrics should not include repartition internals: {metrics:?}"
+            );
+            assert!(
+                !metrics.contains("sent_rows"),
+                "removed repartition counters should not be registered: {metrics:?}"
+            );
+        }
+
+        let internal_metrics = repartition_metric_names(
+            metrics::MetricType::Internal,
+            Partitioning::Hash(vec![col("c0", &schema)?], 2),
+        )
+        .await?;
+        assert!(
+            internal_metrics.contains("hash_compute_time"),
+            "internal metrics should include hash_compute_time: {internal_metrics:?}"
+        );
+        assert!(
+            internal_metrics.contains("route_time"),
+            "internal metrics should include route_time: {internal_metrics:?}"
+        );
+        assert!(
+            internal_metrics.contains("batch_build_time"),
+            "internal metrics should include batch_build_time: {internal_metrics:?}"
+        );
+        assert!(
+            internal_metrics.contains("channel_wait_time"),
+            "internal metrics should include channel_wait_time: {internal_metrics:?}"
+        );
+        assert!(
+            internal_metrics.contains("spill_write_time"),
+            "internal metrics should include spill_write_time: {internal_metrics:?}"
+        );
+        assert!(
+            !internal_metrics.contains("partitioned_rows")
+                && !internal_metrics.contains("partitioned_batches")
+                && !internal_metrics.contains("sent_rows")
+                && !internal_metrics.contains("sent_batches"),
+            "removed repartition counters should not be registered: {internal_metrics:?}"
+        );
+
+        let round_robin_metrics = repartition_metric_names(
+            metrics::MetricType::Internal,
+            Partitioning::RoundRobinBatch(2),
+        )
+        .await?;
+        assert!(
+            !round_robin_metrics.contains("hash_compute_time")
+                && !round_robin_metrics.contains("route_time")
+                && !round_robin_metrics.contains("batch_build_time"),
+            "round-robin repartition should not register hash-only metrics: {round_robin_metrics:?}"
+        );
+        assert!(
+            round_robin_metrics.contains("channel_wait_time")
+                && round_robin_metrics.contains("spill_write_time"),
+            "round-robin repartition should still register non-hash internals: {round_robin_metrics:?}"
+        );
+
+        Ok(())
+    }
+
+    async fn repartition_metric_names(
+        analyze_level: metrics::MetricType,
+        partitioning: Partitioning,
+    ) -> Result<HashSet<String>> {
+        let schema = test_schema();
+        let partition = create_vec_batches(4);
+        let partitions = vec![partition];
+
+        let exec = TestMemoryExec::try_new_exec(&partitions, Arc::clone(&schema), None)?;
+        let exec = RepartitionExec::try_new(exec, partitioning)?;
+
+        let mut session_config = SessionConfig::new();
+        session_config.options_mut().explain.analyze_level = analyze_level;
+        let task_ctx =
+            Arc::new(TaskContext::default().with_session_config(session_config));
+
+        for i in 0..exec.partitioning().partition_count() {
+            let mut stream = exec.execute(i, Arc::clone(&task_ctx))?;
+            while let Some(result) = stream.next().await {
+                result?;
+            }
+        }
+
+        Ok(exec
+            .metrics()
+            .expect("repartition metrics")
+            .iter()
+            .map(|metric| metric.value().name().to_owned())
+            .collect())
+    }
+
     #[tokio::test]
     async fn test_repartition_with_coalescing() -> Result<()> {
         let schema = test_schema();
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto
index 865887d41e111..973c44332ad3a 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -1230,6 +1230,9 @@ message AnalyzeExecNode {
   // Empty means "plan only". Absent (has_metric_categories=false) means "all".
   bool has_metric_categories = 5;
   repeated string metric_categories = 6;
+  // Metric type filters for EXPLAIN ANALYZE output.
+  // Empty preserves the legacy default of summary and dev metrics.
+  repeated string metric_types = 7;
 }
 
 message CrossJoinExecNode {
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs
index b8639afd04a89..d0f38eddaad34 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -981,6 +981,9 @@ impl serde::Serialize for AnalyzeExecNode {
         if !self.metric_categories.is_empty() {
             len += 1;
         }
+        if !self.metric_types.is_empty() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("datafusion.AnalyzeExecNode", len)?;
         if self.verbose {
             struct_ser.serialize_field("verbose", &self.verbose)?;
@@ -1000,6 +1003,9 @@ impl serde::Serialize for AnalyzeExecNode {
         if !self.metric_categories.is_empty() {
             struct_ser.serialize_field("metricCategories", &self.metric_categories)?;
         }
+        if !self.metric_types.is_empty() {
+            struct_ser.serialize_field("metricTypes", &self.metric_types)?;
+        }
         struct_ser.end()
     }
 }
@@ -1019,6 +1025,8 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
             "hasMetricCategories",
             "metric_categories",
             "metricCategories",
+            "metric_types",
+            "metricTypes",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -1029,6 +1037,7 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
             Schema,
             HasMetricCategories,
             MetricCategories,
+            MetricTypes,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -1056,6 +1065,7 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
                             "schema" => Ok(GeneratedField::Schema),
                             "hasMetricCategories" | "has_metric_categories" => Ok(GeneratedField::HasMetricCategories),
                             "metricCategories" | "metric_categories" => Ok(GeneratedField::MetricCategories),
+                            "metricTypes" | "metric_types" => Ok(GeneratedField::MetricTypes),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -1081,6 +1091,7 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
                 let mut schema__ = None;
                 let mut has_metric_categories__ = None;
                 let mut metric_categories__ = None;
+                let mut metric_types__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Verbose => {
@@ -1119,6 +1130,12 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
                             }
                             metric_categories__ = Some(map_.next_value()?);
                         }
+                        GeneratedField::MetricTypes => {
+                            if metric_types__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("metricTypes"));
+                            }
+                            metric_types__ = Some(map_.next_value()?);
+                        }
                     }
                 }
                 Ok(AnalyzeExecNode {
@@ -1128,6 +1145,7 @@ impl<'de> serde::Deserialize<'de> for AnalyzeExecNode {
                     schema: schema__,
                     has_metric_categories: has_metric_categories__.unwrap_or_default(),
                     metric_categories: metric_categories__.unwrap_or_default(),
+                    metric_types: metric_types__.unwrap_or_default(),
                 })
             }
         }
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs
index b742e82ea24ec..f166edb8e7426 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1831,6 +1831,10 @@ pub struct AnalyzeExecNode {
     pub has_metric_categories: bool,
     #[prost(string, repeated, tag = "6")]
     pub metric_categories: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+    /// Metric type filters for EXPLAIN ANALYZE output.
+    /// Empty preserves the legacy default of summary and dev metrics.
+    #[prost(string, repeated, tag = "7")]
+    pub metric_types: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CrossJoinExecNode {
diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index 5172a552fad4f..1bcb753201252 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -1821,10 +1821,19 @@ impl protobuf::PhysicalPlanNode {
         } else {
             None
         };
+        let metric_types = if analyze.metric_types.is_empty() {
+            vec![MetricType::Summary, MetricType::Dev]
+        } else {
+            analyze
+                .metric_types
+                .iter()
+                .map(|s| s.parse::<MetricType>())
+                .collect::<Result<Vec<_>>>()?
+        };
         Ok(Arc::new(AnalyzeExec::new(
             analyze.verbose,
             analyze.show_statistics,
-            vec![MetricType::Summary, MetricType::Dev],
+            metric_types,
             metric_categories,
             input,
             Arc::new(convert_required!(analyze.schema)?),
@@ -2319,6 +2328,11 @@ impl protobuf::PhysicalPlanNode {
                     schema: Some(exec.schema().as_ref().try_into()?),
                     has_metric_categories,
                     metric_categories,
+                    metric_types: exec
+                        .metric_types()
+                        .iter()
+                        .map(|metric_type| metric_type.to_string())
+                        .collect(),
                 },
             ))),
         })
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index fa342ae9079d5..86a500f63b7c9 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -1549,14 +1549,26 @@ fn roundtrip_analyze() -> Result<()> {
     let schema = Schema::new(vec![field_a, field_b]);
     let input = Arc::new(PlaceholderRowExec::new(Arc::new(schema.clone())));
 
-    roundtrip_test(Arc::new(AnalyzeExec::new(
+    let metric_types = vec![MetricType::Summary, MetricType::Dev, MetricType::Internal];
+    let plan = Arc::new(AnalyzeExec::new(
         false,
         false,
-        vec![MetricType::Summary, MetricType::Dev],
+        metric_types.clone(),
         None,
         input,
         Arc::new(schema),
-    )))
+    ));
+
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+    let roundtripped = roundtrip_test_and_return(plan, &ctx, &codec, &proto_converter)?;
+    let analyze = roundtripped
+        .downcast_ref::<AnalyzeExec>()
+        .expect("roundtripped AnalyzeExec");
+    assert_eq!(analyze.metric_types(), metric_types.as_slice());
+
+    Ok(())
 }
 
 #[tokio::test]
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 8396a60137ee1..10670bbe692ed 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -425,7 +425,7 @@ datafusion.execution.target_partitions 7 Number of partitions for query executio
 datafusion.execution.time_zone NULL The default time zone Some functions, e.g. `now` return timestamps in this time zone
 datafusion.execution.use_row_number_estimates_to_optimize_partitioning false Should DataFusion use row number estimates at the input to decide whether increasing parallelism is beneficial or not. By default, only exact row numbers (not estimates) are used for this decision. Setting this flag to `true` will likely produce better plans. if the source of statistics is accurate. We plan to make this the default in the future.
 datafusion.explain.analyze_categories all Which metric categories to include in "EXPLAIN ANALYZE" output. Comma-separated list of: "rows", "bytes", "timing", "uncategorized". Use "none" to show plan structure only, or "all" (default) to show everything. Metrics without a declared category are treated as "uncategorized".
-datafusion.explain.analyze_level dev Verbosity level for "EXPLAIN ANALYZE". Default is "dev" "summary" shows common metrics for high-level insights. "dev" provides deep operator-level introspection for developers.
+datafusion.explain.analyze_level dev Verbosity level for "EXPLAIN ANALYZE". Default is "dev" "summary" shows common metrics for high-level insights. "dev" provides deep operator-level introspection for developers. "internal" provides low-level kernel debugging metrics.
 datafusion.explain.format indent Display format of explain. Default is "indent". When set to "tree", it will print the plan in a tree-rendered format.
 datafusion.explain.logical_plan_only false When set to true, the explain statement will only print logical plans
 datafusion.explain.physical_plan_only false When set to true, the explain statement will only print physical plans
@@ -889,7 +889,7 @@ show functions
 statement ok
 reset datafusion.catalog.information_schema;
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index f7f9426e2bd32..7a61b6b43eb02 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -180,7 +180,7 @@ The following configuration settings are available:
 | datafusion.explain.show_schema                                          | false                     | When set to true, the explain statement will print schema information                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | datafusion.explain.format                                               | indent                    | Display format of explain. Default is "indent". When set to "tree", it will print the plan in a tree-rendered format.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | datafusion.explain.tree_maximum_render_width                            | 240                       | (format=tree only) Maximum total width of the rendered tree. When set to 0, the tree will have no width limit.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
-| datafusion.explain.analyze_level                                        | dev                       | Verbosity level for "EXPLAIN ANALYZE". Default is "dev" "summary" shows common metrics for high-level insights. "dev" provides deep operator-level introspection for developers.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
+| datafusion.explain.analyze_level                                        | dev                       | Verbosity level for "EXPLAIN ANALYZE". Default is "dev" "summary" shows common metrics for high-level insights. "dev" provides deep operator-level introspection for developers. "internal" provides low-level kernel debugging metrics.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 | datafusion.explain.analyze_categories                                   | all                       | Which metric categories to include in "EXPLAIN ANALYZE" output. Comma-separated list of: "rows", "bytes", "timing", "uncategorized". Use "none" to show plan structure only, or "all" (default) to show everything. Metrics without a declared category are treated as "uncategorized".                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | datafusion.sql_parser.parse_float_as_decimal                            | false                     | When set to true, SQL parser will parse float as decimal type                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
 | datafusion.sql_parser.enable_ident_normalization                        | true                      | When set to true, SQL parser will normalize ident (convert ident to lowercase when not quoted)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
diff --git a/docs/source/user-guide/metrics.md b/docs/source/user-guide/metrics.md
index 7e0363f4ceb9b..cbb7b660b2992 100644
--- a/docs/source/user-guide/metrics.md
+++ b/docs/source/user-guide/metrics.md
@@ -42,6 +42,19 @@ DataFusion operators expose runtime metrics so you can understand where time is
 | ----------- | ----------------------------------------------------------------- |
 | selectivity | Selectivity of the filter, calculated as output_rows / input_rows |
 
+### RepartitionExec
+
+| Metric            | Level    | Description                                                                                                    |
+| ----------------- | -------- | -------------------------------------------------------------------------------------------------------------- |
+| fetch_time        | dev      | Time spent polling input partitions for batches.                                                               |
+| repartition_time  | dev      | End-to-end time spent repartitioning input batches, excluding input fetch and output channel send time.        |
+| send_time         | dev      | Time spent preparing and sending partitioned batches to each output partition channel.                         |
+| hash_compute_time | internal | Hash repartition only: time spent evaluating partitioning expressions and computing row hashes.                |
+| route_time        | internal | Hash repartition only: time spent assigning hashed row indices to output partitions.                           |
+| batch_build_time  | internal | Hash repartition only: time spent materializing output batches from routed row indices.                        |
+| channel_wait_time | internal | Time spent waiting for output channel capacity while sending partitioned batches.                              |
+| spill_write_time  | internal | Time spent writing partitioned batches to spill storage when memory reservation for an output channel is full. |
+
 ## TODO
 
 Add metrics for the remaining operators
diff --git a/docs/source/user-guide/sql/explain.md b/docs/source/user-guide/sql/explain.md
index 23101632625b1..3ec851d96f20b 100644
--- a/docs/source/user-guide/sql/explain.md
+++ b/docs/source/user-guide/sql/explain.md
@@ -249,6 +249,6 @@ EXPLAIN ANALYZE SELECT SUM(x) FROM table GROUP BY b;
 
 By default `EXPLAIN ANALYZE` shows the aggregated metrics from all partitions for each operator. If you need to display per-partition metrics, use `EXPLAIN ANALYZE VERBOSE`.
 
-You can also set `datafusion.explain.analyze_level` from the [configuration value] to control the detail level for the metrics displayed.
+You can also set `datafusion.explain.analyze_level` from the [configuration value] to control the detail level for the metrics displayed. The levels are `summary`, `dev`, and `internal`; `internal` includes low-level kernel debugging metrics intended for DataFusion development.
 
 [configuration value]: ../configs.md