Display all partitions and files in EXPLAIN VERBOSE (#6711)

Adds DisplayAs trait for structs which could show more details when
formatted in the verbose mode
Resolves #6383

Co-authored-by: Andrew Lamb <[email protected]>

Author: qrilka

benchmarks/src/bin/tpch.rs

@@ -303,14 +303,14 @@ async fn execute_query(
     if debug {
         println!(
             "=== Physical plan ===\n{}\n",
-            displayable(physical_plan.as_ref()).indent()
+            displayable(physical_plan.as_ref()).indent(true)
         );
     }
     let result = collect(physical_plan.clone(), state.task_ctx()).await?;
     if debug {
         println!(
             "=== Physical plan with metrics ===\n{}\n",
-            DisplayableExecutionPlan::with_metrics(physical_plan.as_ref()).indent()
+            DisplayableExecutionPlan::with_metrics(physical_plan.as_ref()).indent(true)
         );
         if !result.is_empty() {
             // do not call print_batches if there are no batches as the result is confusing

datafusion/core/src/datasource/file_format/csv.rs

@@ -20,7 +20,7 @@
 use std::any::Any;
 use std::collections::HashSet;
 use std::fmt;
-use std::fmt::{Debug, Display};
+use std::fmt::Debug;
 use std::sync::Arc;
 
 use arrow::csv::WriterBuilder;
@@ -51,7 +51,7 @@ use crate::datasource::physical_plan::{
 use crate::error::Result;
 use crate::execution::context::SessionState;
 use crate::physical_plan::insert::{DataSink, InsertExec};
-use crate::physical_plan::Statistics;
+use crate::physical_plan::{DisplayAs, DisplayFormatType, Statistics};
 use crate::physical_plan::{ExecutionPlan, SendableRecordBatchStream};
 
 /// The default file extension of csv files
@@ -443,14 +443,19 @@ impl Debug for CsvSink {
     }
 }
 
-impl Display for CsvSink {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(
-            f,
-            "CsvSink(writer_mode={:?}, file_groups={})",
-            self.config.writer_mode,
-            FileGroupDisplay(&self.config.file_groups),
-        )
+impl DisplayAs for CsvSink {
+    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match t {
+            DisplayFormatType::Default | DisplayFormatType::Verbose => {
+                write!(
+                    f,
+                    "CsvSink(writer_mode={:?}, file_groups=",
+                    self.config.writer_mode
+                )?;
+                FileGroupDisplay(&self.config.file_groups).fmt_as(t, f)?;
+                write!(f, ")")
+            }
+        }
     }
 }
 

datafusion/core/src/datasource/memory.rs

@@ -20,7 +20,7 @@
 use futures::StreamExt;
 use log::debug;
 use std::any::Any;
-use std::fmt::{self, Debug, Display};
+use std::fmt::{self, Debug};
 use std::sync::Arc;
 
 use arrow::datatypes::SchemaRef;
@@ -36,9 +36,9 @@ use crate::logical_expr::Expr;
 use crate::physical_plan::common::AbortOnDropSingle;
 use crate::physical_plan::insert::{DataSink, InsertExec};
 use crate::physical_plan::memory::MemoryExec;
-use crate::physical_plan::ExecutionPlan;
 use crate::physical_plan::{common, SendableRecordBatchStream};
 use crate::physical_plan::{repartition::RepartitionExec, Partitioning};
+use crate::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan};
 
 /// Type alias for partition data
 pub type PartitionData = Arc<RwLock<Vec<RecordBatch>>>;
@@ -213,10 +213,14 @@ impl Debug for MemSink {
     }
 }
 
-impl Display for MemSink {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let partition_count = self.batches.len();
-        write!(f, "MemoryTable (partitions={partition_count})")
+impl DisplayAs for MemSink {
+    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match t {
+            DisplayFormatType::Default | DisplayFormatType::Verbose => {
+                let partition_count = self.batches.len();
+                write!(f, "MemoryTable (partitions={partition_count})")
+            }
+        }
     }
 }
 

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

@@ -22,7 +22,7 @@ use crate::datasource::physical_plan::{
 use crate::error::Result;
 use crate::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use crate::physical_plan::{
-    ordering_equivalence_properties_helper, DisplayFormatType, ExecutionPlan,
+    ordering_equivalence_properties_helper, DisplayAs, DisplayFormatType, ExecutionPlan,
     Partitioning, SendableRecordBatchStream,
 };
 use arrow_schema::SchemaRef;
@@ -137,11 +137,8 @@ impl ExecutionPlan for ArrowExec {
         t: DisplayFormatType,
         f: &mut std::fmt::Formatter,
     ) -> std::fmt::Result {
-        match t {
-            DisplayFormatType::Default => {
-                write!(f, "ArrowExec: {}", self.base_config)
-            }
-        }
+        write!(f, "ArrowExec: ")?;
+        self.base_config.fmt_as(t, f)
     }
 
     fn statistics(&self) -> Statistics {

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

@@ -20,7 +20,7 @@ use crate::error::Result;
 use crate::physical_plan::expressions::PhysicalSortExpr;
 use crate::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use crate::physical_plan::{
-    ordering_equivalence_properties_helper, DisplayFormatType, ExecutionPlan,
+    ordering_equivalence_properties_helper, DisplayAs, DisplayFormatType, ExecutionPlan,
     Partitioning, SendableRecordBatchStream, Statistics,
 };
 use datafusion_execution::TaskContext;
@@ -146,11 +146,8 @@ impl ExecutionPlan for AvroExec {
         t: DisplayFormatType,
         f: &mut std::fmt::Formatter,
     ) -> std::fmt::Result {
-        match t {
-            DisplayFormatType::Default => {
-                write!(f, "AvroExec: {}", self.base_config)
-            }
-        }
+        write!(f, "AvroExec: ")?;
+        self.base_config.fmt_as(t, f)
     }
 
     fn statistics(&self) -> Statistics {

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

@@ -27,7 +27,7 @@ use crate::physical_plan::common::AbortOnDropSingle;
 use crate::physical_plan::expressions::PhysicalSortExpr;
 use crate::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use crate::physical_plan::{
-    ordering_equivalence_properties_helper, DisplayFormatType, ExecutionPlan,
+    ordering_equivalence_properties_helper, DisplayAs, DisplayFormatType, ExecutionPlan,
     Partitioning, SendableRecordBatchStream, Statistics,
 };
 use arrow::csv;
@@ -177,15 +177,9 @@ impl ExecutionPlan for CsvExec {
         t: DisplayFormatType,
         f: &mut std::fmt::Formatter,
     ) -> std::fmt::Result {
-        match t {
-            DisplayFormatType::Default => {
-                write!(
-                    f,
-                    "CsvExec: {}, has_header={}",
-                    self.base_config, self.has_header,
-                )
-            }
-        }
+        write!(f, "CsvExec: ")?;
+        self.base_config.fmt_as(t, f)?;
+        write!(f, ", has_header={}", self.has_header)
     }
 
     fn statistics(&self) -> Statistics {

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

@@ -26,7 +26,7 @@ use crate::physical_plan::common::AbortOnDropSingle;
 use crate::physical_plan::expressions::PhysicalSortExpr;
 use crate::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use crate::physical_plan::{
-    ordering_equivalence_properties_helper, DisplayFormatType, ExecutionPlan,
+    ordering_equivalence_properties_helper, DisplayAs, DisplayFormatType, ExecutionPlan,
     Partitioning, SendableRecordBatchStream, Statistics,
 };
 use datafusion_execution::TaskContext;
@@ -155,11 +155,8 @@ impl ExecutionPlan for NdJsonExec {
         t: DisplayFormatType,
         f: &mut std::fmt::Formatter,
     ) -> std::fmt::Result {
-        match t {
-            DisplayFormatType::Default => {
-                write!(f, "JsonExec: {}", self.base_config)
-            }
-        }
+        write!(f, "JsonExec: ")?;
+        self.base_config.fmt_as(t, f)
     }
 
     fn statistics(&self) -> Statistics {