datafusion_physical_plan/

display.rs

// 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.

//! Implementation of physical plan display. See
//! [`crate::displayable`] for examples of how to format

use std::collections::{BTreeMap, HashMap};
use std::fmt;
use std::fmt::Formatter;
use std::time::Duration;

use arrow::datatypes::SchemaRef;

use datafusion_common::display::{GraphvizBuilder, PlanType, StringifiedPlan};
use datafusion_expr::display_schema;
use datafusion_physical_expr::LexOrdering;

use crate::metrics::{MetricCategory, MetricType, MetricValue};
use crate::render_tree::RenderTree;

use crate::statistics::{StatisticsArgs, StatisticsContext};

use super::{ExecutionPlan, ExecutionPlanVisitor, accept};

/// Options for controlling how each [`ExecutionPlan`] should format itself
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum DisplayFormatType {
    /// Default, compact format. Example: `FilterExec: c12 < 10.0`
    ///
    /// This format is designed to provide a detailed textual description
    /// of all parts of the plan.
    Default,
    /// Verbose, showing all available details.
    ///
    /// This form is even more detailed than [`Self::Default`]
    Verbose,
    /// TreeRender, displayed in the `tree` explain type.
    ///
    /// This format is inspired by DuckDB's explain plans. The information
    /// presented should be "user friendly", and contain only the most relevant
    /// information for understanding a plan. It should NOT contain the same level
    /// of detail information as the  [`Self::Default`] format.
    ///
    /// In this mode, each line has one of two formats:
    ///
    /// 1. A string without a `=`, which is printed in its own line
    ///
    /// 2. A string with a `=` that is treated as a `key=value pair`. Everything
    ///    before the first `=` is treated as the key, and everything after the
    ///    first `=` is treated as the value.
    ///
    /// For example, if the output of `TreeRender` is this:
    /// ```text
    /// Parquet
    /// partition_sizes=[1]
    /// ```
    ///
    /// It is rendered in the center of a box in the following way:
    ///
    /// ```text
    /// ┌───────────────────────────┐
    /// │       DataSourceExec      │
    /// │    --------------------   │
    /// │    partition_sizes: [1]   │
    /// │          Parquet          │
    /// └───────────────────────────┘
    /// ```
    TreeRender,
}

/// Wraps an `ExecutionPlan` with various methods for formatting
///
///
/// # Example
/// ```
/// # use std::sync::Arc;
/// # use arrow::datatypes::{Field, Schema, DataType};
/// # use datafusion_expr::Operator;
/// # use datafusion_physical_expr::expressions::{binary, col, lit};
/// # use datafusion_physical_plan::{displayable, ExecutionPlan};
/// # use datafusion_physical_plan::empty::EmptyExec;
/// # use datafusion_physical_plan::filter::FilterExec;
/// # let schema = Schema::new(vec![Field::new("i", DataType::Int32, false)]);
/// # let plan = EmptyExec::new(Arc::new(schema));
/// # let i = col("i", &plan.schema()).unwrap();
/// # let predicate = binary(i, Operator::Eq, lit(1), &plan.schema()).unwrap();
/// # let plan: Arc<dyn ExecutionPlan> = Arc::new(FilterExec::try_new(predicate, Arc::new(plan)).unwrap());
/// // Get a one line description (Displayable)
/// let display_plan = displayable(plan.as_ref());
///
/// // you can use the returned objects to format plans
/// // where you can use `Display` such as  format! or println!
/// assert_eq!(
///    &format!("The plan is: {}", display_plan.one_line()),
///   "The plan is: FilterExec: i@0 = 1\n"
/// );
/// // You can also print out the plan and its children in indented mode
/// assert_eq!(display_plan.indent(false).to_string(),
///   "FilterExec: i@0 = 1\
///   \n  EmptyExec\
///   \n"
/// );
/// ```
#[derive(Debug, Clone)]
pub struct DisplayableExecutionPlan<'a> {
    inner: &'a dyn ExecutionPlan,
    /// How to show metrics
    show_metrics: ShowMetrics,
    /// If statistics should be displayed
    show_statistics: bool,
    /// If schema should be displayed. See [`Self::set_show_schema`]
    show_schema: bool,
    /// Which metric categories should be included when rendering
    metric_types: Vec<MetricType>,
    /// Optional filter by semantic category (rows / bytes / timing).
    /// `None` means show all categories; `Some(vec![])` means plan-only.
    metric_categories: Option<Vec<MetricCategory>>,
    /// Optional filter by metric names. Only metric names in this list
    /// will be rendered.
    metric_names: Option<Vec<String>>,
    // (TreeRender) Maximum total width of the rendered tree
    tree_maximum_render_width: usize,
    /// Optional summary totals (currently only used by `pgjson`) — the total
    /// row count and wall-clock duration of the `AnalyzeExec` execution.
    summary: Option<AnalyzeSummary>,
}

/// Summary information attached to the root of an `EXPLAIN ANALYZE`
/// pgjson render.
#[derive(Debug, Clone, Copy)]
struct AnalyzeSummary {
    total_rows: Option<usize>,
    duration: Option<Duration>,
}

impl<'a> DisplayableExecutionPlan<'a> {
    fn default_metric_types() -> Vec<MetricType> {
        vec![MetricType::Summary, MetricType::Dev]
    }

    /// Create a wrapper around an [`ExecutionPlan`] which can be
    /// pretty printed in a variety of ways
    pub fn new(inner: &'a dyn ExecutionPlan) -> Self {
        Self {
            inner,
            show_metrics: ShowMetrics::None,
            show_statistics: false,
            show_schema: false,
            metric_types: Self::default_metric_types(),
            metric_categories: None,
            metric_names: None,
            tree_maximum_render_width: 240,
            summary: None,
        }
    }

    /// Create a wrapper around an [`ExecutionPlan`] which can be
    /// pretty printed in a variety of ways that also shows aggregated
    /// metrics
    pub fn with_metrics(inner: &'a dyn ExecutionPlan) -> Self {
        Self {
            inner,
            show_metrics: ShowMetrics::Aggregated,
            show_statistics: false,
            show_schema: false,
            metric_types: Self::default_metric_types(),
            metric_categories: None,
            metric_names: None,
            tree_maximum_render_width: 240,
            summary: None,
        }
    }

    /// Create a wrapper around an [`ExecutionPlan`] which can be
    /// pretty printed in a variety of ways that also shows all low
    /// level metrics
    pub fn with_full_metrics(inner: &'a dyn ExecutionPlan) -> Self {
        Self {
            inner,
            show_metrics: ShowMetrics::Full,
            show_statistics: false,
            show_schema: false,
            metric_types: Self::default_metric_types(),
            metric_categories: None,
            metric_names: None,
            tree_maximum_render_width: 240,
            summary: None,
        }
    }

    /// Enable display of schema
    ///
    /// If true, plans will be displayed with schema information at the end
    /// of each line. The format is `schema=[[a:Int32;N, b:Int32;N, c:Int32;N]]`
    pub fn set_show_schema(mut self, show_schema: bool) -> Self {
        self.show_schema = show_schema;
        self
    }

    /// Enable display of statistics
    pub fn set_show_statistics(mut self, show_statistics: bool) -> Self {
        self.show_statistics = show_statistics;
        self
    }

    /// Specify which metric types should be rendered alongside the plan
    pub fn set_metric_types(mut self, metric_types: Vec<MetricType>) -> Self {
        self.metric_types = metric_types;
        self
    }

    /// Specify which metric categories to include.
    ///
    /// - `None` means show all categories (default).
    /// - `Some(vec![])` means plan-only — suppress all metrics.
    /// - `Some(vec![Rows])` means show only row-count metrics (plus
    ///   uncategorized metrics).
    ///
    /// See [`MetricCategory`] for the determinism properties of each
    /// category.
    pub fn set_metric_categories(
        mut self,
        metric_categories: Option<Vec<MetricCategory>>,
    ) -> Self {
        self.metric_categories = metric_categories;
        self
    }

    /// Specify which metric names to include.
    ///
    /// - An empty vector means plan-only — suppress all metrics.
    /// - `vec!["metric_1"]` means show only the metric named `metric_1`.
    ///
    /// Name filtering is intersected with other types of filters, like metric
    /// category and metric type.
    pub fn set_metric_names(mut self, metric_names: Vec<String>) -> Self {
        self.metric_names = Some(metric_names);
        self
    }

    /// Set the maximum render width for the tree format
    pub fn set_tree_maximum_render_width(mut self, width: usize) -> Self {
        self.tree_maximum_render_width = width;
        self
    }

    /// Attach an `EXPLAIN ANALYZE` summary (total output rows and duration)
    /// to the rendered output. Currently only used by [`Self::pgjson`], which
    /// serializes the summary alongside the root plan object.
    pub fn set_summary(
        mut self,
        total_rows: Option<usize>,
        duration: Option<Duration>,
    ) -> Self {
        self.summary = Some(AnalyzeSummary {
            total_rows,
            duration,
        });
        self
    }

    /// Return a `format`able structure that produces a single line
    /// per node.
    ///
    /// ```text
    /// ProjectionExec: expr=[a]
    ///   CoalesceBatchesExec: target_batch_size=8192
    ///     FilterExec: a < 5
    ///       RepartitionExec: partitioning=RoundRobinBatch(16)
    ///         DataSourceExec: source=...",
    /// ```
    pub fn indent(&self, verbose: bool) -> impl fmt::Display + 'a {
        let format_type = if verbose {
            DisplayFormatType::Verbose
        } else {
            DisplayFormatType::Default
        };
        struct Wrapper<'a> {
            format_type: DisplayFormatType,
            plan: &'a dyn ExecutionPlan,
            show_metrics: ShowMetrics,
            show_statistics: bool,
            show_schema: bool,
            metric_types: Vec<MetricType>,
            metric_categories: Option<Vec<MetricCategory>>,
            metric_names: Option<Vec<String>>,
        }
        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter) -> fmt::Result {
                let mut visitor = IndentVisitor {
                    t: self.format_type,
                    f,
                    indent: 0,
                    show_metrics: self.show_metrics,
                    show_statistics: self.show_statistics,
                    show_schema: self.show_schema,
                    metric_types: &self.metric_types,
                    metric_categories: self.metric_categories.as_deref(),
                    metric_names: self.metric_names.as_deref(),
                };
                accept(self.plan, &mut visitor)
            }
        }
        Wrapper {
            format_type,
            plan: self.inner,
            show_metrics: self.show_metrics,
            show_statistics: self.show_statistics,
            show_schema: self.show_schema,
            metric_types: self.metric_types.clone(),
            metric_categories: self.metric_categories.clone(),
            metric_names: self.metric_names.clone(),
        }
    }

    /// Returns a `format`able structure that produces graphviz format for execution plan, which can
    /// be directly visualized [here](https://dreampuf.github.io/GraphvizOnline).
    ///
    /// An example is
    /// ```dot
    /// strict digraph dot_plan {
    //     0[label="ProjectionExec: expr=[id@0 + 2 as employee.id + Int32(2)]",tooltip=""]
    //     1[label="EmptyExec",tooltip=""]
    //     0 -> 1
    // }
    /// ```
    pub fn graphviz(&self) -> impl fmt::Display + 'a {
        struct Wrapper<'a> {
            plan: &'a dyn ExecutionPlan,
            show_metrics: ShowMetrics,
            show_statistics: bool,
            metric_types: Vec<MetricType>,
            metric_categories: Option<Vec<MetricCategory>>,
            metric_names: Option<Vec<String>>,
        }
        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter) -> fmt::Result {
                let t = DisplayFormatType::Default;

                let mut visitor = GraphvizVisitor {
                    f,
                    t,
                    show_metrics: self.show_metrics,
                    show_statistics: self.show_statistics,
                    metric_types: &self.metric_types,
                    metric_categories: self.metric_categories.as_deref(),
                    metric_names: self.metric_names.as_deref(),
                    graphviz_builder: GraphvizBuilder::default(),
                    parents: Vec::new(),
                };

                visitor.start_graph()?;

                accept(self.plan, &mut visitor)?;

                visitor.end_graph()?;
                Ok(())
            }
        }

        Wrapper {
            plan: self.inner,
            show_metrics: self.show_metrics,
            show_statistics: self.show_statistics,
            metric_types: self.metric_types.clone(),
            metric_categories: self.metric_categories.clone(),
            metric_names: self.metric_names.clone(),
        }
    }

    /// Formats the plan using a ASCII art like tree
    ///
    /// See [`DisplayFormatType::TreeRender`] for more details.
    pub fn tree_render(&self) -> impl fmt::Display + 'a {
        struct Wrapper<'a> {
            plan: &'a dyn ExecutionPlan,
            maximum_render_width: usize,
        }
        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter) -> fmt::Result {
                let mut visitor = TreeRenderVisitor {
                    f,
                    maximum_render_width: self.maximum_render_width,
                };
                visitor.visit(self.plan)
            }
        }
        Wrapper {
            plan: self.inner,
            maximum_render_width: self.tree_maximum_render_width,
        }
    }

    /// Returns a `format`able structure that produces PostgreSQL-style JSON
    /// output, mirroring the logical-plan pgjson format.
    ///
    /// Each node is rendered as a JSON object with:
    /// - `"Node Type"` — `ExecutionPlan::name()`
    /// - `"Details"` — the one-line `DisplayAs::Default` rendering
    /// - `"Output"` — schema column names (when `set_show_schema(true)`)
    /// - `"Actual Rows"` / `"Actual Total Time"` — PG-canonical metric keys
    ///   populated from `output_rows` / `elapsed_compute` when available
    /// - `"Extras"` — remaining metrics keyed by DataFusion metric name
    /// - `"Plans"` — array of child nodes
    ///
    /// When a summary has been set via [`Self::set_summary`], `"Total Rows"`
    /// and `"Duration"` fields are attached at the root.
    pub fn pgjson(&self, verbose: bool) -> impl fmt::Display + 'a {
        struct Wrapper<'a> {
            plan: &'a dyn ExecutionPlan,
            verbose: bool,
            show_metrics: ShowMetrics,
            show_schema: bool,
            metric_types: Vec<MetricType>,
            metric_categories: Option<Vec<MetricCategory>>,
            metric_names: Option<Vec<String>>,
            summary: Option<AnalyzeSummary>,
        }
        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter) -> fmt::Result {
                let mut visitor = PgJsonExecutionPlanVisitor {
                    verbose: self.verbose,
                    show_metrics: self.show_metrics,
                    show_schema: self.show_schema,
                    metric_types: &self.metric_types,
                    metric_categories: self.metric_categories.as_deref(),
                    metric_names: self.metric_names.as_deref(),
                    objects: HashMap::new(),
                    parent_ids: Vec::new(),
                    next_id: 0,
                    root: None,
                };
                accept(self.plan, &mut visitor).map_err(|_| fmt::Error)?;
                let root = visitor.root.ok_or(fmt::Error)?;
                let mut root_entry = serde_json::json!({ "Plan": root });
                if let Some(summary) = self.summary {
                    if let Some(total_rows) = summary.total_rows {
                        root_entry["Total Rows"] = serde_json::Value::from(total_rows);
                    }
                    if let Some(duration) = summary.duration {
                        root_entry["Duration"] =
                            serde_json::Value::from(format!("{duration:?}"));
                    }
                }
                let doc = serde_json::Value::Array(vec![root_entry]);
                write!(
                    f,
                    "{}",
                    serde_json::to_string_pretty(&doc).map_err(|_| fmt::Error)?
                )
            }
        }

        Wrapper {
            plan: self.inner,
            verbose,
            show_metrics: self.show_metrics,
            show_schema: self.show_schema,
            metric_types: self.metric_types.clone(),
            metric_categories: self.metric_categories.clone(),
            metric_names: self.metric_names.clone(),
            summary: self.summary,
        }
    }

    /// Return a single-line summary of the root of the plan
    /// Example: `ProjectionExec: expr=[a@0 as a]`.
    pub fn one_line(&self) -> impl fmt::Display + 'a {
        struct Wrapper<'a> {
            plan: &'a dyn ExecutionPlan,
            show_metrics: ShowMetrics,
            show_statistics: bool,
            show_schema: bool,
            metric_types: Vec<MetricType>,
            metric_categories: Option<Vec<MetricCategory>>,
            metric_names: Option<Vec<String>>,
        }

        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter) -> fmt::Result {
                let mut visitor = IndentVisitor {
                    f,
                    t: DisplayFormatType::Default,
                    indent: 0,
                    show_metrics: self.show_metrics,
                    show_statistics: self.show_statistics,
                    show_schema: self.show_schema,
                    metric_types: &self.metric_types,
                    metric_categories: self.metric_categories.as_deref(),
                    metric_names: self.metric_names.as_deref(),
                };
                visitor.pre_visit(self.plan)?;
                Ok(())
            }
        }

        Wrapper {
            plan: self.inner,
            show_metrics: self.show_metrics,
            show_statistics: self.show_statistics,
            show_schema: self.show_schema,
            metric_types: self.metric_types.clone(),
            metric_categories: self.metric_categories.clone(),
            metric_names: self.metric_names.clone(),
        }
    }

    #[deprecated(since = "47.0.0", note = "indent() or tree_render() instead")]
    pub fn to_stringified(
        &self,
        verbose: bool,
        plan_type: PlanType,
        explain_format: DisplayFormatType,
    ) -> StringifiedPlan {
        match (&explain_format, &plan_type) {
            (DisplayFormatType::TreeRender, PlanType::FinalPhysicalPlan) => {
                StringifiedPlan::new(plan_type, self.tree_render().to_string())
            }
            _ => StringifiedPlan::new(plan_type, self.indent(verbose).to_string()),
        }
    }
}

/// Enum representing the different levels of metrics to display
#[derive(Debug, Clone, Copy)]
enum ShowMetrics {
    /// Do not show any metrics
    None,

    /// Show aggregated metrics across partition
    Aggregated,

    /// Show full per-partition metrics
    Full,
}

/// Formats plans with a single line per node.
///
/// # Example
///
/// ```text
/// ProjectionExec: expr=[column1@0 + 2 as column1 + Int64(2)]
///   FilterExec: column1@0 = 5
///     ValuesExec
/// ```
struct IndentVisitor<'a, 'b> {
    /// How to format each node
    t: DisplayFormatType,
    /// Write to this formatter
    f: &'a mut Formatter<'b>,
    /// Indent size
    indent: usize,
    /// How to show metrics
    show_metrics: ShowMetrics,
    /// If statistics should be displayed
    show_statistics: bool,
    /// If schema should be displayed
    show_schema: bool,
    /// Which metric types should be rendered
    metric_types: &'a [MetricType],
    /// Optional filter by semantic category (rows / bytes / timing).
    metric_categories: Option<&'a [MetricCategory]>,
    /// Optional filter by metric name.
    metric_names: Option<&'a [String]>,
}

impl ExecutionPlanVisitor for IndentVisitor<'_, '_> {
    type Error = fmt::Error;
    fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        write!(self.f, "{:indent$}", "", indent = self.indent * 2)?;
        plan.fmt_as(self.t, self.f)?;
        match self.show_metrics {
            ShowMetrics::None => {}
            ShowMetrics::Aggregated => {
                if let Some(metrics) = plan.metrics() {
                    let mut metrics = metrics
                        .filter_by_metric_types(self.metric_types)
                        .aggregate_by_name()
                        .sorted_for_display()
                        .timestamps_removed();
                    if let Some(cats) = self.metric_categories {
                        metrics = metrics.filter_by_categories(cats);
                    }
                    if let Some(names) = self.metric_names {
                        metrics = metrics.filter_by_names(names);
                    }
                    write!(self.f, ", metrics=[{metrics}]")?;
                } else {
                    write!(self.f, ", metrics=[]")?;
                }
            }
            ShowMetrics::Full => {
                if let Some(metrics) = plan.metrics() {
                    let mut metrics = metrics.filter_by_metric_types(self.metric_types);
                    if let Some(cats) = self.metric_categories {
                        metrics = metrics.filter_by_categories(cats);
                    }
                    if let Some(names) = self.metric_names {
                        metrics = metrics.filter_by_names(names);
                    }
                    write!(self.f, ", metrics=[{metrics}]")?;
                } else {
                    write!(self.f, ", metrics=[]")?;
                }
            }
        }
        if self.show_statistics {
            let stats = StatisticsContext::new()
                .compute(plan, &StatisticsArgs::new())
                .map_err(|_e| fmt::Error)?;
            write!(self.f, ", statistics=[{stats}]")?;
        }
        if self.show_schema {
            write!(
                self.f,
                ", schema={}",
                display_schema(plan.schema().as_ref())
            )?;
        }
        writeln!(self.f)?;
        self.indent += 1;
        Ok(true)
    }

    fn post_visit(&mut self, _plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        self.indent -= 1;
        Ok(true)
    }
}

struct GraphvizVisitor<'a, 'b> {
    f: &'a mut Formatter<'b>,
    /// How to format each node
    t: DisplayFormatType,
    /// How to show metrics
    show_metrics: ShowMetrics,
    /// If statistics should be displayed
    show_statistics: bool,
    /// Which metric types should be rendered
    metric_types: &'a [MetricType],
    /// Optional filter by semantic category
    metric_categories: Option<&'a [MetricCategory]>,
    /// Optional filter by metric name.
    metric_names: Option<&'a [String]>,

    graphviz_builder: GraphvizBuilder,
    /// Used to record parent node ids when visiting a plan.
    parents: Vec<usize>,
}

impl GraphvizVisitor<'_, '_> {
    fn start_graph(&mut self) -> fmt::Result {
        self.graphviz_builder.start_graph(self.f)
    }

    fn end_graph(&mut self) -> fmt::Result {
        self.graphviz_builder.end_graph(self.f)
    }
}

impl ExecutionPlanVisitor for GraphvizVisitor<'_, '_> {
    type Error = fmt::Error;

    fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        let id = self.graphviz_builder.next_id();

        struct Wrapper<'a>(&'a dyn ExecutionPlan, DisplayFormatType);

        impl fmt::Display for Wrapper<'_> {
            fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
                self.0.fmt_as(self.1, f)
            }
        }

        let label = { format!("{}", Wrapper(plan, self.t)) };

        let metrics = match self.show_metrics {
            ShowMetrics::None => "".to_string(),
            ShowMetrics::Aggregated => {
                if let Some(metrics) = plan.metrics() {
                    let mut metrics = metrics
                        .filter_by_metric_types(self.metric_types)
                        .aggregate_by_name()
                        .sorted_for_display()
                        .timestamps_removed();
                    if let Some(cats) = self.metric_categories {
                        metrics = metrics.filter_by_categories(cats);
                    }
                    if let Some(names) = self.metric_names {
                        metrics = metrics.filter_by_names(names);
                    }
                    format!("metrics=[{metrics}]")
                } else {
                    "metrics=[]".to_string()
                }
            }
            ShowMetrics::Full => {
                if let Some(metrics) = plan.metrics() {
                    let mut metrics = metrics.filter_by_metric_types(self.metric_types);
                    if let Some(cats) = self.metric_categories {
                        metrics = metrics.filter_by_categories(cats);
                    }
                    if let Some(names) = self.metric_names {
                        metrics = metrics.filter_by_names(names);
                    }
                    format!("metrics=[{metrics}]")
                } else {
                    "metrics=[]".to_string()
                }
            }
        };

        let statistics = if self.show_statistics {
            let stats = StatisticsContext::new()
                .compute(plan, &StatisticsArgs::new())
                .map_err(|_e| fmt::Error)?;
            format!("statistics=[{stats}]")
        } else {
            "".to_string()
        };

        let delimiter = if !metrics.is_empty() && !statistics.is_empty() {
            ", "
        } else {
            ""
        };

        self.graphviz_builder.add_node(
            self.f,
            id,
            &label,
            Some(&format!("{metrics}{delimiter}{statistics}")),
        )?;

        if let Some(parent_node_id) = self.parents.last() {
            self.graphviz_builder
                .add_edge(self.f, *parent_node_id, id)?;
        }

        self.parents.push(id);

        Ok(true)
    }

    fn post_visit(&mut self, _plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        self.parents.pop();
        Ok(true)
    }
}

/// Formats physical plans into PostgreSQL-style JSON output with live
/// per-operator metrics.
///
/// This visitor mirrors the logical-plan `PgJsonVisitor` in
/// `datafusion-expr`: during `pre_visit` it assembles a JSON object for the
/// current node; during `post_visit` it attaches that object into its
/// parent's `"Plans"` array (or stores it as the root).
struct PgJsonExecutionPlanVisitor<'a> {
    verbose: bool,
    show_metrics: ShowMetrics,
    show_schema: bool,
    metric_types: &'a [MetricType],
    metric_categories: Option<&'a [MetricCategory]>,
    metric_names: Option<&'a [String]>,
    objects: HashMap<u32, serde_json::Value>,
    parent_ids: Vec<u32>,
    next_id: u32,
    root: Option<serde_json::Value>,
}

impl PgJsonExecutionPlanVisitor<'_> {
    /// Produce the one-line `DisplayAs::Default` rendering of a node.
    fn one_line_details(plan: &dyn ExecutionPlan) -> String {
        struct One<'b>(&'b dyn ExecutionPlan);
        impl fmt::Display for One<'_> {
            fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
                self.0.fmt_as(DisplayFormatType::Default, f)
            }
        }
        // Some operators include internal newlines; collapse them so the
        // rendered JSON value stays on a single line.
        format!("{}", One(plan))
            .replace('\n', " ")
            .trim()
            .to_string()
    }

    /// Render the given `MetricValue` into the most natural `serde_json::Value`
    /// we can produce: a number for simple counts/gauges/times, a float-ms for
    /// `ElapsedCompute`, and a string fallback for anything else.
    fn metric_value_to_json(value: &MetricValue) -> serde_json::Value {
        match value {
            MetricValue::OutputRows(c) => serde_json::Value::from(c.value()),
            MetricValue::SpillCount(c)
            | MetricValue::OutputBatches(c)
            | MetricValue::SpilledRows(c) => serde_json::Value::from(c.value()),
            MetricValue::SpilledBytes(c) | MetricValue::OutputBytes(c) => {
                serde_json::Value::from(c.value())
            }
            MetricValue::CurrentMemoryUsage(g) => serde_json::Value::from(g.value()),
            MetricValue::ElapsedCompute(t) => {
                // Emit as float milliseconds to align with PG's
                // `"Actual Total Time"` convention. DataFusion tracks compute
                // time (summed across partitions), not wall time — visualizers
                // should be read with that caveat in mind.
                let ms = (t.value() as f64) / 1_000_000.0;
                serde_json::Value::from(ms)
            }
            MetricValue::Count { count, .. } => serde_json::Value::from(count.value()),
            MetricValue::Gauge { gauge, .. } => serde_json::Value::from(gauge.value()),
            MetricValue::PeakMemoryUsage { gauge, .. } => {
                serde_json::Value::from(gauge.value())
            }
            MetricValue::Time { time, .. } => {
                let ms = (time.value() as f64) / 1_000_000.0;
                serde_json::Value::from(ms)
            }
            // Timestamps, PruningMetrics, Ratio, Custom: fall back to Display.
            other => serde_json::Value::String(format!("{other}")),
        }
    }

    /// Populate `"Actual Rows"`, `"Actual Total Time"`, and `"Extras"` for
    /// the given node from its aggregated `MetricsSet`, honoring the same
    /// filtering pipeline used by `IndentVisitor`.
    fn attach_metrics(&self, plan: &dyn ExecutionPlan, object: &mut serde_json::Value) {
        if matches!(self.show_metrics, ShowMetrics::None) {
            return;
        }
        let Some(metrics) = plan.metrics() else {
            return;
        };

        let metrics = match self.show_metrics {
            ShowMetrics::None => return,
            ShowMetrics::Aggregated => metrics
                .filter_by_metric_types(self.metric_types)
                .aggregate_by_name()
                .sorted_for_display()
                .timestamps_removed(),
            ShowMetrics::Full => metrics.filter_by_metric_types(self.metric_types),
        };
        let metrics = if let Some(cats) = self.metric_categories {
            metrics.filter_by_categories(cats)
        } else {
            metrics
        };

        let metrics = if let Some(names) = self.metric_names {
            metrics.filter_by_names(names)
        } else {
            metrics
        };

        // Build the Extras bucket, while extracting PG-canonical keys to the
        // top level.
        let mut extras = serde_json::Map::new();
        for metric in metrics.iter() {
            let value = metric.value();
            match value {
                MetricValue::OutputRows(c) => {
                    object["Actual Rows"] = serde_json::Value::from(c.value());
                }
                MetricValue::ElapsedCompute(t) => {
                    let ms = (t.value() as f64) / 1_000_000.0;
                    object["Actual Total Time"] = serde_json::Value::from(ms);
                }
                _ => {
                    extras.insert(
                        value.name().to_string(),
                        Self::metric_value_to_json(value),
                    );
                }
            }
        }
        if !extras.is_empty() {
            object["Extras"] = serde_json::Value::Object(extras);
        }
    }
}

impl ExecutionPlanVisitor for PgJsonExecutionPlanVisitor<'_> {
    type Error = fmt::Error;

    fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        let id = self.next_id;
        self.next_id += 1;

        // Build fields in reading order: Node Type, Details, (schema),
        // (metrics), Plans last — so the JSON output reads top-down like a
        // PostgreSQL plan.
        let mut object = serde_json::json!({
            "Node Type": plan.name(),
            "Details": Self::one_line_details(plan),
        });

        if self.show_schema || self.verbose {
            // Always include output columns when a caller asked for schema;
            // also include them in verbose mode so the pgjson output mirrors
            // the extra context shown by indent's verbose flag.
            let columns: Vec<serde_json::Value> = plan
                .schema()
                .fields()
                .iter()
                .map(|f| serde_json::Value::String(f.name().to_string()))
                .collect();
            object["Output"] = serde_json::Value::Array(columns);
        }

        self.attach_metrics(plan, &mut object);

        object["Plans"] = serde_json::Value::Array(vec![]);

        self.objects.insert(id, object);
        self.parent_ids.push(id);
        Ok(true)
    }

    fn post_visit(&mut self, _plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
        let id = self.parent_ids.pop().ok_or(fmt::Error)?;
        let current = self.objects.remove(&id).ok_or(fmt::Error)?;

        if let Some(parent_id) = self.parent_ids.last() {
            let parent = self.objects.get_mut(parent_id).ok_or(fmt::Error)?;
            let plans = parent
                .get_mut("Plans")
                .and_then(|p| p.as_array_mut())
                .ok_or(fmt::Error)?;
            plans.push(current);
        } else {
            self.root = Some(current);
        }
        Ok(true)
    }
}

/// This module implements a tree-like art renderer for execution plans,
/// based on DuckDB's implementation:
/// <https://github.com/duckdb/duckdb/blob/main/src/include/duckdb/common/tree_renderer/text_tree_renderer.hpp>
///
/// The rendered output looks like this:
/// ```text
/// ┌───────────────────────────┐
/// │    CoalesceBatchesExec    │
/// └─────────────┬─────────────┘
/// ┌─────────────┴─────────────┐
/// │        HashJoinExec       ├──────────────┐
/// └─────────────┬─────────────┘              │
/// ┌─────────────┴─────────────┐┌─────────────┴─────────────┐
/// │       DataSourceExec      ││       DataSourceExec      │
/// └───────────────────────────┘└───────────────────────────┘
/// ```
///
/// The renderer uses a three-layer approach for each node:
/// 1. Top layer: renders the top borders and connections
/// 2. Content layer: renders the node content and vertical connections
/// 3. Bottom layer: renders the bottom borders and connections
///
/// Each node is rendered in a box of fixed width (NODE_RENDER_WIDTH).
struct TreeRenderVisitor<'a, 'b> {
    /// Write to this formatter
    f: &'a mut Formatter<'b>,
    /// Maximum total width of the rendered tree
    maximum_render_width: usize,
}

impl TreeRenderVisitor<'_, '_> {
    // Unicode box-drawing characters for creating borders and connections.
    const LTCORNER: &'static str = "┌"; // Left top corner
    const RTCORNER: &'static str = "┐"; // Right top corner
    const LDCORNER: &'static str = "└"; // Left bottom corner
    const RDCORNER: &'static str = "┘"; // Right bottom corner

    const TMIDDLE: &'static str = "┬"; // Top T-junction (connects down)
    const LMIDDLE: &'static str = "├"; // Left T-junction (connects right)
    const DMIDDLE: &'static str = "┴"; // Bottom T-junction (connects up)

    const VERTICAL: &'static str = "│"; // Vertical line
    const HORIZONTAL: &'static str = "─"; // Horizontal line

    // TODO: Make these variables configurable.
    const NODE_RENDER_WIDTH: usize = 29; // Width of each node's box
    const MAX_EXTRA_LINES: usize = 30; // Maximum number of extra info lines per node

    /// Main entry point for rendering an execution plan as a tree.
    /// The rendering process happens in three stages for each level of the tree:
    /// 1. Render top borders and connections
    /// 2. Render node content and vertical connections
    /// 3. Render bottom borders and connections
    pub fn visit(&mut self, plan: &dyn ExecutionPlan) -> Result<(), fmt::Error> {
        let root = RenderTree::create_tree(plan);

        for y in 0..root.height {
            // Start by rendering the top layer.
            self.render_top_layer(&root, y)?;
            // Now we render the content of the boxes
            self.render_box_content(&root, y)?;
            // Render the bottom layer of each of the boxes
            self.render_bottom_layer(&root, y)?;
        }

        Ok(())
    }

    /// Renders the top layer of boxes at the given y-level of the tree.
    /// This includes:
    /// - Top corners (┌─┐) for nodes
    /// - Horizontal connections between nodes
    /// - Vertical connections to parent nodes
    fn render_top_layer(
        &mut self,
        root: &RenderTree,
        y: usize,
    ) -> Result<(), fmt::Error> {
        for x in 0..root.width {
            if self.maximum_render_width > 0
                && x * Self::NODE_RENDER_WIDTH >= self.maximum_render_width
            {
                break;
            }

            if root.has_node(x, y) {
                write!(self.f, "{}", Self::LTCORNER)?;
                write!(
                    self.f,
                    "{}",
                    Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2 - 1)
                )?;
                if y == 0 {
                    // top level node: no node above this one
                    write!(self.f, "{}", Self::HORIZONTAL)?;
                } else {
                    // render connection to node above this one
                    write!(self.f, "{}", Self::DMIDDLE)?;
                }
                write!(
                    self.f,
                    "{}",
                    Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2 - 1)
                )?;
                write!(self.f, "{}", Self::RTCORNER)?;
            } else {
                let mut has_adjacent_nodes = false;
                for i in 0..(root.width - x) {
                    has_adjacent_nodes = has_adjacent_nodes || root.has_node(x + i, y);
                }
                if !has_adjacent_nodes {
                    // There are no nodes to the right side of this position
                    // no need to fill the empty space
                    continue;
                }
                // there are nodes next to this, fill the space
                write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH))?;
            }
        }
        writeln!(self.f)?;

        Ok(())
    }

    /// Renders the content layer of boxes at the given y-level of the tree.
    /// This includes:
    /// - Node names and extra information
    /// - Vertical borders (│) for boxes
    /// - Vertical connections between nodes
    fn render_box_content(
        &mut self,
        root: &RenderTree,
        y: usize,
    ) -> Result<(), fmt::Error> {
        let mut extra_info: Vec<Vec<String>> = vec![vec![]; root.width];
        let mut extra_height = 0;

        for (x, extra_info_item) in extra_info.iter_mut().enumerate().take(root.width) {
            if let Some(node) = root.get_node(x, y) {
                Self::split_up_extra_info(
                    &node.extra_text,
                    extra_info_item,
                    Self::MAX_EXTRA_LINES,
                );
                if extra_info_item.len() > extra_height {
                    extra_height = extra_info_item.len();
                }
            }
        }

        let halfway_point = extra_height.div_ceil(2);

        // Render the actual node.
        for render_y in 0..=extra_height {
            for (x, _) in root.nodes.iter().enumerate().take(root.width) {
                if self.maximum_render_width > 0
                    && x * Self::NODE_RENDER_WIDTH >= self.maximum_render_width
                {
                    break;
                }

                let mut has_adjacent_nodes = false;
                for i in 0..(root.width - x) {
                    has_adjacent_nodes = has_adjacent_nodes || root.has_node(x + i, y);
                }

                if let Some(node) = root.get_node(x, y) {
                    write!(self.f, "{}", Self::VERTICAL)?;

                    // Figure out what to render.
                    let mut render_text = if render_y == 0 {
                        node.name.clone()
                    } else if render_y <= extra_info[x].len() {
                        extra_info[x][render_y - 1].clone()
                    } else {
                        String::new()
                    };

                    render_text = Self::adjust_text_for_rendering(
                        &render_text,
                        Self::NODE_RENDER_WIDTH - 2,
                    );
                    write!(self.f, "{render_text}")?;

                    if render_y == halfway_point && node.child_positions.len() > 1 {
                        write!(self.f, "{}", Self::LMIDDLE)?;
                    } else {
                        write!(self.f, "{}", Self::VERTICAL)?;
                    }
                } else if render_y == halfway_point {
                    let has_child_to_the_right =
                        Self::should_render_whitespace(root, x, y);
                    if root.has_node(x, y + 1) {
                        // Node right below this one.
                        write!(
                            self.f,
                            "{}",
                            Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2)
                        )?;
                        if has_child_to_the_right {
                            write!(self.f, "{}", Self::TMIDDLE)?;
                            // Have another child to the right, Keep rendering the line.
                            write!(
                                self.f,
                                "{}",
                                Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2)
                            )?;
                        } else {
                            write!(self.f, "{}", Self::RTCORNER)?;
                            if has_adjacent_nodes {
                                // Only a child below this one: fill the reset with spaces.
                                write!(
                                    self.f,
                                    "{}",
                                    " ".repeat(Self::NODE_RENDER_WIDTH / 2)
                                )?;
                            }
                        }
                    } else if has_child_to_the_right {
                        // Child to the right, but no child right below this one: render a full
                        // line.
                        write!(
                            self.f,
                            "{}",
                            Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH)
                        )?;
                    } else if has_adjacent_nodes {
                        // Empty spot: render spaces.
                        write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH))?;
                    }
                } else if render_y >= halfway_point {
                    if root.has_node(x, y + 1) {
                        // Have a node below this empty spot: render a vertical line.
                        write!(
                            self.f,
                            "{}{}",
                            " ".repeat(Self::NODE_RENDER_WIDTH / 2),
                            Self::VERTICAL
                        )?;
                        if has_adjacent_nodes
                            || Self::should_render_whitespace(root, x, y)
                        {
                            write!(
                                self.f,
                                "{}",
                                " ".repeat(Self::NODE_RENDER_WIDTH / 2)
                            )?;
                        }
                    } else if has_adjacent_nodes
                        || Self::should_render_whitespace(root, x, y)
                    {
                        // Empty spot: render spaces.
                        write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH))?;
                    }
                } else if has_adjacent_nodes {
                    // Empty spot: render spaces.
                    write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH))?;
                }
            }
            writeln!(self.f)?;
        }

        Ok(())
    }

    /// Renders the bottom layer of boxes at the given y-level of the tree.
    /// This includes:
    /// - Bottom corners (└─┘) for nodes
    /// - Horizontal connections between nodes
    /// - Vertical connections to child nodes
    fn render_bottom_layer(
        &mut self,
        root: &RenderTree,
        y: usize,
    ) -> Result<(), fmt::Error> {
        for x in 0..=root.width {
            if self.maximum_render_width > 0
                && x * Self::NODE_RENDER_WIDTH >= self.maximum_render_width
            {
                break;
            }
            let mut has_adjacent_nodes = false;
            for i in 0..(root.width - x) {
                has_adjacent_nodes = has_adjacent_nodes || root.has_node(x + i, y);
            }
            if root.get_node(x, y).is_some() {
                write!(self.f, "{}", Self::LDCORNER)?;
                write!(
                    self.f,
                    "{}",
                    Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2 - 1)
                )?;
                if root.has_node(x, y + 1) {
                    // node below this one: connect to that one
                    write!(self.f, "{}", Self::TMIDDLE)?;
                } else {
                    // no node below this one: end the box
                    write!(self.f, "{}", Self::HORIZONTAL)?;
                }
                write!(
                    self.f,
                    "{}",
                    Self::HORIZONTAL.repeat(Self::NODE_RENDER_WIDTH / 2 - 1)
                )?;
                write!(self.f, "{}", Self::RDCORNER)?;
            } else if root.has_node(x, y + 1) {
                write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH / 2))?;
                write!(self.f, "{}", Self::VERTICAL)?;
                if has_adjacent_nodes || Self::should_render_whitespace(root, x, y) {
                    write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH / 2))?;
                }
            } else if has_adjacent_nodes || Self::should_render_whitespace(root, x, y) {
                write!(self.f, "{}", " ".repeat(Self::NODE_RENDER_WIDTH))?;
            }
        }
        writeln!(self.f)?;

        Ok(())
    }

    fn extra_info_separator() -> String {
        "-".repeat(Self::NODE_RENDER_WIDTH - 9)
    }

    fn remove_padding(s: &str) -> String {
        s.trim().to_string()
    }

    pub fn split_up_extra_info(
        extra_info: &HashMap<String, String>,
        result: &mut Vec<String>,
        max_lines: usize,
    ) {
        if extra_info.is_empty() {
            return;
        }

        result.push(Self::extra_info_separator());

        let mut requires_padding = false;
        let mut was_inlined = false;

        // use BTreeMap for repeatable key order
        let sorted_extra_info: BTreeMap<_, _> = extra_info.iter().collect();
        for (key, value) in sorted_extra_info {
            let mut str = Self::remove_padding(value);
            let mut is_inlined = false;
            let available_width = Self::NODE_RENDER_WIDTH - 7;
            let total_size = key.len() + str.len() + 2;
            let is_multiline = str.contains('\n');

            if str.is_empty() {
                str = key.to_string();
            } else if !is_multiline && total_size < available_width {
                str = format!("{key}: {str}");
                is_inlined = true;
            } else {
                str = format!("{key}:\n{str}");
            }

            if is_inlined && was_inlined {
                requires_padding = false;
            }

            if requires_padding {
                result.push(String::new());
            }

            let mut splits: Vec<String> = str.split('\n').map(String::from).collect();
            if splits.len() > max_lines {
                let mut truncated_splits = Vec::new();
                for split in splits.iter().take(max_lines / 2) {
                    truncated_splits.push(split.clone());
                }
                truncated_splits.push("...".to_string());
                for split in splits.iter().skip(splits.len() - max_lines / 2) {
                    truncated_splits.push(split.clone());
                }
                splits = truncated_splits;
            }
            for split in splits {
                Self::split_string_buffer(&split, result);
            }
            if result.len() > max_lines {
                result.truncate(max_lines);
                result.push("...".to_string());
            }

            requires_padding = true;
            was_inlined = is_inlined;
        }
    }

    /// Adjusts text to fit within the specified width by:
    /// 1. Truncating with ellipsis if too long
    /// 2. Center-aligning within the available space if shorter
    fn adjust_text_for_rendering(source: &str, max_render_width: usize) -> String {
        let render_width = source.chars().count();
        if render_width > max_render_width {
            let truncated = &source[..max_render_width - 3];
            format!("{truncated}...")
        } else {
            let total_spaces = max_render_width - render_width;
            let half_spaces = total_spaces / 2;
            let extra_left_space = if total_spaces.is_multiple_of(2) { 0 } else { 1 };
            format!(
                "{}{}{}",
                " ".repeat(half_spaces + extra_left_space),
                source,
                " ".repeat(half_spaces)
            )
        }
    }

    /// Determines if whitespace should be rendered at a given position.
    /// This is important for:
    /// 1. Maintaining proper spacing between sibling nodes
    /// 2. Ensuring correct alignment of connections between parents and children
    /// 3. Preserving the tree structure's visual clarity
    fn should_render_whitespace(root: &RenderTree, x: usize, y: usize) -> bool {
        let mut found_children = 0;

        for i in (0..=x).rev() {
            let node = root.get_node(i, y);
            if root.has_node(i, y + 1) {
                found_children += 1;
            }
            if let Some(node) = node {
                if node.child_positions.len() > 1
                    && found_children < node.child_positions.len()
                {
                    return true;
                }

                return false;
            }
        }

        false
    }

    fn split_string_buffer(source: &str, result: &mut Vec<String>) {
        let mut character_pos = 0;
        let mut start_pos = 0;
        let mut render_width = 0;
        let mut last_possible_split = 0;

        let chars: Vec<char> = source.chars().collect();

        while character_pos < chars.len() {
            // Treating each char as width 1 for simplification
            let char_width = 1;

            // Does the next character make us exceed the line length?
            if render_width + char_width > Self::NODE_RENDER_WIDTH - 2 {
                if start_pos + 8 > last_possible_split {
                    // The last character we can split on is one of the first 8 characters of the line
                    // to not create very small lines we instead split on the current character
                    last_possible_split = character_pos;
                }

                result.push(source[start_pos..last_possible_split].to_string());
                render_width = character_pos - last_possible_split;
                start_pos = last_possible_split;
                character_pos = last_possible_split;
            }

            // check if we can split on this character
            if Self::can_split_on_this_char(chars[character_pos]) {
                last_possible_split = character_pos;
            }

            character_pos += 1;
            render_width += char_width;
        }

        if source.len() > start_pos {
            // append the remainder of the input
            result.push(source[start_pos..].to_string());
        }
    }

    fn can_split_on_this_char(c: char) -> bool {
        (!c.is_ascii_digit() && !c.is_ascii_uppercase() && !c.is_ascii_lowercase())
            && c != '_'
    }
}

/// Trait for types which could have additional details when formatted in `Verbose` mode
pub trait DisplayAs {
    /// Format according to `DisplayFormatType`, used when verbose representation looks
    /// different from the default one
    ///
    /// Should not include a newline
    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> fmt::Result;
}

/// A new type wrapper to display `T` implementing`DisplayAs` using the `Default` mode
pub struct DefaultDisplay<T>(pub T);

impl<T: DisplayAs> fmt::Display for DefaultDisplay<T> {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        self.0.fmt_as(DisplayFormatType::Default, f)
    }
}

/// A new type wrapper to display `T` implementing `DisplayAs` using the `Verbose` mode
pub struct VerboseDisplay<T>(pub T);

impl<T: DisplayAs> fmt::Display for VerboseDisplay<T> {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        self.0.fmt_as(DisplayFormatType::Verbose, f)
    }
}

/// A wrapper to customize partitioned file display
#[derive(Debug)]
pub struct ProjectSchemaDisplay<'a>(pub &'a SchemaRef);

impl fmt::Display for ProjectSchemaDisplay<'_> {
    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
        let parts: Vec<_> = self
            .0
            .fields()
            .iter()
            .map(|x| x.name().to_owned())
            .collect::<Vec<String>>();
        write!(f, "[{}]", parts.join(", "))
    }
}

pub fn display_orderings(f: &mut Formatter, orderings: &[LexOrdering]) -> fmt::Result {
    if !orderings.is_empty() {
        let start = if orderings.len() == 1 {
            ", output_ordering="
        } else {
            ", output_orderings=["
        };
        write!(f, "{start}")?;
        for (idx, ordering) in orderings.iter().enumerate() {
            match idx {
                0 => write!(f, "[{ordering}]")?,
                _ => write!(f, ", [{ordering}]")?,
            }
        }
        let end = if orderings.len() == 1 { "" } else { "]" };
        write!(f, "{end}")?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::fmt::Write;
    use std::sync::Arc;

    use datafusion_common::{
        Result, Statistics, internal_datafusion_err, tree_node::TreeNodeRecursion,
    };
    use datafusion_execution::{SendableRecordBatchStream, TaskContext};
    use datafusion_physical_expr::PhysicalExpr;

    use crate::statistics::StatisticsArgs;
    use crate::{
        ChildrenPropertiesMode, DisplayAs, ExecutionPlan, PlanProperties,
        ReplaceChildrenOptions,
    };

    use super::DisplayableExecutionPlan;

    #[derive(Debug, Clone, Copy)]
    enum TestStatsExecPlan {
        Panic,
        Error,
        Ok,
    }

    impl DisplayAs for TestStatsExecPlan {
        fn fmt_as(
            &self,
            _t: crate::DisplayFormatType,
            f: &mut std::fmt::Formatter,
        ) -> std::fmt::Result {
            write!(f, "TestStatsExecPlan")
        }
    }

    impl ExecutionPlan for TestStatsExecPlan {
        fn name(&self) -> &'static str {
            "TestStatsExecPlan"
        }

        fn properties(&self) -> &Arc<PlanProperties> {
            unimplemented!()
        }

        fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
            vec![]
        }

        fn replace_children(
            self: Arc<Self>,
            _: Vec<Arc<dyn ExecutionPlan>>,
            _: ReplaceChildrenOptions,
        ) -> Result<Arc<dyn ExecutionPlan>> {
            unimplemented!()
        }

        fn apply_expressions(
            &self,
            _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
        ) -> Result<TreeNodeRecursion> {
            Ok(TreeNodeRecursion::Continue)
        }

        fn with_new_children(
            self: Arc<Self>,
            children: Vec<Arc<dyn ExecutionPlan>>,
        ) -> Result<Arc<dyn ExecutionPlan>> {
            self.replace_children(
                children,
                ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
            )
        }

        fn execute(
            &self,
            _: usize,
            _: Arc<TaskContext>,
        ) -> Result<SendableRecordBatchStream> {
            todo!()
        }

        fn statistics_from_inputs(
            &self,
            _input_stats: &[Arc<Statistics>],
            args: &StatisticsArgs,
        ) -> Result<Arc<Statistics>> {
            if args.partition().is_some() {
                return Ok(Arc::new(Statistics::new_unknown(self.schema().as_ref())));
            }
            match self {
                Self::Panic => panic!("expected panic"),
                Self::Error => Err(internal_datafusion_err!("expected error")),
                Self::Ok => Ok(Arc::new(Statistics::new_unknown(self.schema().as_ref()))),
            }
        }
    }

    fn test_stats_display(exec: TestStatsExecPlan, show_stats: bool) {
        let display =
            DisplayableExecutionPlan::new(&exec).set_show_statistics(show_stats);

        let mut buf = String::new();
        write!(&mut buf, "{}", display.one_line()).unwrap();
        let buf = buf.trim();
        assert_eq!(buf, "TestStatsExecPlan");
    }

    #[test]
    fn test_display_when_stats_panic_with_no_show_stats() {
        test_stats_display(TestStatsExecPlan::Panic, false);
    }

    #[test]
    fn test_display_when_stats_error_with_no_show_stats() {
        test_stats_display(TestStatsExecPlan::Error, false);
    }

    #[test]
    fn test_display_when_stats_ok_with_no_show_stats() {
        test_stats_display(TestStatsExecPlan::Ok, false);
    }

    #[test]
    #[should_panic(expected = "expected panic")]
    fn test_display_when_stats_panic_with_show_stats() {
        test_stats_display(TestStatsExecPlan::Panic, true);
    }

    #[test]
    #[should_panic(expected = "Error")] // fmt::Error
    fn test_display_when_stats_error_with_show_stats() {
        test_stats_display(TestStatsExecPlan::Error, true);
    }

    #[test]
    fn test_display_when_stats_ok_with_show_stats() {
        test_stats_display(TestStatsExecPlan::Ok, false);
    }

    mod pgjson {
        use std::sync::Arc;
        use std::time::Duration;

        use arrow::datatypes::{DataType, Field, Schema};
        use insta::assert_snapshot;

        use super::super::DisplayableExecutionPlan;
        use crate::empty::EmptyExec;
        use crate::filter::FilterExec;
        use crate::projection::ProjectionExec;
        use crate::{ChildrenPropertiesMode, ExecutionPlan, ReplaceChildrenOptions};
        use datafusion_physical_expr::expressions::{binary, col, lit};
        use datafusion_physical_expr::{Partitioning, PhysicalExpr};

        fn sample_plan() -> Arc<dyn ExecutionPlan> {
            let schema = Arc::new(Schema::new(vec![
                Field::new("a", DataType::Int32, false),
                Field::new("b", DataType::Int32, false),
            ]));
            let empty = Arc::new(EmptyExec::new(Arc::clone(&schema)));
            let predicate = binary(
                col("a", &schema).unwrap(),
                datafusion_expr::Operator::Gt,
                lit(5i32),
                &schema,
            )
            .unwrap();
            let filter = Arc::new(FilterExec::try_new(predicate, empty).unwrap());
            let proj_expr: Vec<(Arc<dyn PhysicalExpr>, String)> =
                vec![(col("a", &schema).unwrap(), "a".to_string())];
            let _ = Partitioning::UnknownPartitioning(1);
            Arc::new(ProjectionExec::try_new(proj_expr, filter).unwrap())
        }

        #[test]
        fn pgjson_renders_plan_without_metrics() {
            let plan = sample_plan();
            let out = DisplayableExecutionPlan::new(plan.as_ref())
                .pgjson(false)
                .to_string();
            let value: serde_json::Value = serde_json::from_str(&out).unwrap();
            // Root is an array with one {"Plan": ...} entry.
            let root = value
                .as_array()
                .expect("root array")
                .first()
                .expect("root entry")
                .get("Plan")
                .expect("plan object");
            assert_eq!(root["Node Type"].as_str(), Some("ProjectionExec"));
            assert!(root.get("Actual Rows").is_none());
            assert!(root.get("Extras").is_none());
            let plans = root["Plans"].as_array().expect("Plans array");
            assert_eq!(plans.len(), 1);
            assert_eq!(plans[0]["Node Type"].as_str(), Some("FilterExec"));
        }

        #[test]
        fn pgjson_emits_pg_canonical_metric_keys() {
            use crate::metrics::{Count, Metric, MetricValue, MetricsSet, Time};
            use crate::{DisplayFormatType, ExecutionPlan, PlanProperties};
            use datafusion_common::Result;
            use datafusion_execution::{SendableRecordBatchStream, TaskContext};

            // Wrap `sample_plan()` with an adapter node that exposes a
            // hand-crafted `MetricsSet` so we can assert the PG key mapping
            // without running anything.
            #[derive(Debug)]
            struct WithMetrics {
                inner: Arc<dyn ExecutionPlan>,
                metrics: MetricsSet,
            }
            impl crate::DisplayAs for WithMetrics {
                fn fmt_as(
                    &self,
                    _t: DisplayFormatType,
                    f: &mut std::fmt::Formatter,
                ) -> std::fmt::Result {
                    write!(f, "WithMetrics")
                }
            }
            impl ExecutionPlan for WithMetrics {
                fn name(&self) -> &'static str {
                    "WithMetrics"
                }
                fn properties(&self) -> &Arc<PlanProperties> {
                    self.inner.properties()
                }
                fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
                    vec![&self.inner]
                }
                fn apply_expressions(
                    &self,
                    _f: &mut dyn FnMut(
                        &Arc<dyn PhysicalExpr>,
                    ) -> Result<
                        datafusion_common::tree_node::TreeNodeRecursion,
                    >,
                ) -> Result<datafusion_common::tree_node::TreeNodeRecursion>
                {
                    Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
                }

                fn replace_children(
                    self: Arc<Self>,
                    _: Vec<Arc<dyn ExecutionPlan>>,
                    _: ReplaceChildrenOptions,
                ) -> Result<Arc<dyn ExecutionPlan>> {
                    unimplemented!()
                }
                fn with_new_children(
                    self: Arc<Self>,
                    children: Vec<Arc<dyn ExecutionPlan>>,
                ) -> Result<Arc<dyn ExecutionPlan>> {
                    self.replace_children(
                        children,
                        ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
                    )
                }
                fn execute(
                    &self,
                    _: usize,
                    _: Arc<TaskContext>,
                ) -> Result<SendableRecordBatchStream> {
                    unimplemented!()
                }
                fn metrics(&self) -> Option<MetricsSet> {
                    Some(self.metrics.clone())
                }
            }

            let mut metrics = MetricsSet::new();
            let rows = Count::new();
            rows.add(42);
            metrics.push(Arc::new(Metric::new(MetricValue::OutputRows(rows), None)));
            let elapsed = Time::new();
            elapsed.add_duration(Duration::from_millis(5));
            metrics.push(Arc::new(Metric::new(
                MetricValue::ElapsedCompute(elapsed),
                None,
            )));
            let batches = Count::new();
            batches.add(7);
            metrics.push(Arc::new(Metric::new(
                MetricValue::OutputBatches(batches),
                None,
            )));

            let plan: Arc<dyn ExecutionPlan> = Arc::new(WithMetrics {
                inner: sample_plan(),
                metrics,
            });

            let out = DisplayableExecutionPlan::with_metrics(plan.as_ref())
                .pgjson(false)
                .to_string();
            let value: serde_json::Value = serde_json::from_str(&out).unwrap();
            let root = value[0].get("Plan").expect("plan");
            assert_eq!(root["Actual Rows"].as_u64(), Some(42));
            assert_eq!(root["Actual Total Time"].as_f64(), Some(5.0));
            assert_eq!(root["Extras"]["output_batches"].as_u64(), Some(7));

            let metric_names = vec!["output_rows".to_string()];
            for rendered in [
                DisplayableExecutionPlan::with_metrics(plan.as_ref())
                    .set_metric_names(metric_names.clone())
                    .indent(false)
                    .to_string(),
                DisplayableExecutionPlan::with_full_metrics(plan.as_ref())
                    .set_metric_names(metric_names.clone())
                    .indent(false)
                    .to_string(),
                DisplayableExecutionPlan::with_metrics(plan.as_ref())
                    .set_metric_names(metric_names.clone())
                    .graphviz()
                    .to_string(),
                DisplayableExecutionPlan::with_full_metrics(plan.as_ref())
                    .set_metric_names(metric_names.clone())
                    .graphviz()
                    .to_string(),
            ] {
                assert!(rendered.contains("output_rows"));
                assert!(!rendered.contains("elapsed_compute"));
                assert!(!rendered.contains("output_batches"));
            }

            let out = DisplayableExecutionPlan::with_metrics(plan.as_ref())
                .set_metric_names(metric_names)
                .pgjson(false)
                .to_string();
            let value: serde_json::Value = serde_json::from_str(&out).unwrap();
            let root = value[0].get("Plan").expect("plan");
            assert_eq!(root["Actual Rows"].as_u64(), Some(42));
            assert!(root.get("Actual Total Time").is_none());
            assert!(root.get("Extras").is_none());
        }

        #[test]
        fn pgjson_includes_summary_when_set() {
            let plan = sample_plan();
            let out = DisplayableExecutionPlan::with_metrics(plan.as_ref())
                .set_summary(Some(42), Some(Duration::from_millis(7)))
                .pgjson(false)
                .to_string();
            let value: serde_json::Value = serde_json::from_str(&out).unwrap();
            let entry = &value.as_array().unwrap()[0];
            assert_eq!(entry["Total Rows"].as_u64(), Some(42));
            assert!(entry["Duration"].is_string());
        }

        #[test]
        fn pgjson_snapshot_of_sample_plan() {
            let plan = sample_plan();
            let out = DisplayableExecutionPlan::new(plan.as_ref())
                .pgjson(false)
                .to_string();
            // This snapshot assumes `serde_json` is built with the
            // `preserve_order` feature (enabled via this crate's dev-deps).
            assert_snapshot!(out, @r#"
            [
              {
                "Plan": {
                  "Node Type": "ProjectionExec",
                  "Details": "ProjectionExec: expr=[a@0 as a]",
                  "Plans": [
                    {
                      "Node Type": "FilterExec",
                      "Details": "FilterExec: a@0 > 5",
                      "Plans": [
                        {
                          "Node Type": "EmptyExec",
                          "Details": "EmptyExec",
                          "Plans": []
                        }
                      ]
                    }
                  ]
                }
              }
            ]
            "#);
        }
    }
}