Simplify the printing of all plans containing expr in tree mode (#15249)

* Simplify the printing of all plans containing `expr` in `tree` mode

* Update datafusion/physical-expr-common/src/sort_expr.rs

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

* fix

---------

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

Author: irenjj

datafusion/datasource-parquet/src/source.rs

@@ -34,6 +34,7 @@ use datafusion_common::config::TableParquetOptions;
 use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_optimizer::pruning::PruningPredicate;
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder};
@@ -580,7 +581,7 @@ impl FileSource for ParquetSource {
             }
             DisplayFormatType::TreeRender => {
                 if let Some(predicate) = self.predicate() {
-                    writeln!(f, "predicate={predicate}")?;
+                    writeln!(f, "predicate={}", fmt_sql(predicate.as_ref()))?;
                 }
                 Ok(())
             }

datafusion/physical-expr-common/src/sort_expr.rs

@@ -17,7 +17,7 @@
 
 //! Sort expressions
 
-use crate::physical_expr::PhysicalExpr;
+use crate::physical_expr::{fmt_sql, PhysicalExpr};
 use std::fmt;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
@@ -117,6 +117,16 @@ impl PhysicalSortExpr {
         self.options.nulls_first = false;
         self
     }
+
+    /// Like [`PhysicalExpr::fmt_sql`] prints a [`PhysicalSortExpr`] in a SQL-like format.
+    pub fn fmt_sql(&self, f: &mut Formatter) -> fmt::Result {
+        write!(
+            f,
+            "{} {}",
+            fmt_sql(self.expr.as_ref()),
+            to_str(&self.options)
+        )
+    }
 }
 
 /// Access the PhysicalSortExpr as a PhysicalExpr

datafusion/physical-plan/src/aggregates/mod.rs

@@ -830,6 +830,7 @@ impl DisplayAs for AggregateExec {
                         })
                         .collect()
                 };
+                // TODO: Implement `fmt_sql` for `AggregateFunctionExpr`.
                 let a: Vec<String> = self
                     .aggr_expr
                     .iter()

datafusion/physical-plan/src/filter.rs

@@ -54,6 +54,7 @@ use datafusion_physical_expr::{
     ExprBoundaries, PhysicalExpr,
 };
 
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
@@ -330,7 +331,7 @@ impl DisplayAs for FilterExec {
                 write!(f, "FilterExec: {}{}", self.predicate, display_projections)
             }
             DisplayFormatType::TreeRender => {
-                write!(f, "predicate={}", self.predicate)
+                write!(f, "predicate={}", fmt_sql(self.predicate.as_ref()))
             }
         }
     }

datafusion/physical-plan/src/joins/hash_join.rs

@@ -82,6 +82,7 @@ use datafusion_physical_expr::PhysicalExprRef;
 use datafusion_physical_expr_common::datum::compare_op_for_nested;
 
 use ahash::RandomState;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use futures::{ready, Stream, StreamExt, TryStreamExt};
 use parking_lot::Mutex;
 
@@ -672,7 +673,9 @@ impl DisplayAs for HashJoinExec {
                 let on = self
                     .on
                     .iter()
-                    .map(|(c1, c2)| format!("({} = {})", c1, c2))
+                    .map(|(c1, c2)| {
+                        format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+                    })
                     .collect::<Vec<String>>()
                     .join(", ");
 

datafusion/physical-plan/src/joins/sort_merge_join.rs

@@ -70,6 +70,7 @@ use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_execution::TaskContext;
 use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use datafusion_physical_expr::PhysicalExprRef;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement};
 
 use futures::{Stream, StreamExt};
@@ -373,7 +374,9 @@ impl DisplayAs for SortMergeJoinExec {
                 let on = self
                     .on
                     .iter()
-                    .map(|(c1, c2)| format!("({} = {})", c1, c2))
+                    .map(|(c1, c2)| {
+                        format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+                    })
                     .collect::<Vec<String>>()
                     .join(", ");
 

datafusion/physical-plan/src/joins/symmetric_hash_join.rs

@@ -74,6 +74,7 @@ use datafusion_expr::interval_arithmetic::Interval;
 use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
 use datafusion_physical_expr::PhysicalExprRef;
+use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement};
 
 use ahash::RandomState;
@@ -384,7 +385,9 @@ impl DisplayAs for SymmetricHashJoinExec {
                 let on = self
                     .on
                     .iter()
-                    .map(|(c1, c2)| format!("({} = {})", c1, c2))
+                    .map(|(c1, c2)| {
+                        format!("({} = {})", fmt_sql(c1.as_ref()), fmt_sql(c2.as_ref()))
+                    })
                     .collect::<Vec<String>>()
                     .join(", ");
 

datafusion/physical-plan/src/sorts/sort_preserving_merge.rs

@@ -187,11 +187,9 @@ impl DisplayAs for SortPreservingMergeExec {
             }
             DisplayFormatType::TreeRender => {
                 for (i, e) in self.expr().iter().enumerate() {
-                    let e = e.to_string();
-                    if i == self.expr().len() - 1 {
-                        writeln!(f, "{e}")?;
-                    } else {
-                        write!(f, "{e}, ")?;
+                    e.fmt_sql(f)?;
+                    if i != self.expr().len() - 1 {
+                        write!(f, ", ")?;
                     }
                 }
                 if let Some(fetch) = self.fetch {