Add name method to execution plan

datafusion-examples/examples/custom_datasource.rs

@@ -226,6 +226,10 @@ impl DisplayAs for CustomExec {
 }
 
 impl ExecutionPlan for CustomExec {
+    fn name(&self) -> &'static str {
+        "CustomExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -122,6 +122,10 @@ impl DisplayAs for ArrowExec {
 }
 
 impl ExecutionPlan for ArrowExec {
+    fn name(&self) -> &'static str {
+        "ArrowExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -99,6 +99,10 @@ impl DisplayAs for AvroExec {
 }
 
 impl ExecutionPlan for AvroExec {
+    fn name(&self) -> &'static str {
+        "AvroExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -160,6 +160,10 @@ impl DisplayAs for CsvExec {
 }
 
 impl ExecutionPlan for CsvExec {
+    fn name(&self) -> &'static str {
+        "CsvExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

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

@@ -127,6 +127,10 @@ impl DisplayAs for NdJsonExec {
 }
 
 impl ExecutionPlan for NdJsonExec {
+    fn name(&self) -> &'static str {
+        "NdJsonExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -315,6 +315,10 @@ impl DisplayAs for ParquetExec {
 }
 
 impl ExecutionPlan for ParquetExec {
+    fn name(&self) -> &'static str {
+        "ParquetExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -1369,6 +1369,10 @@ pub(crate) mod tests {
     }
 
     impl ExecutionPlan for SortRequiredExec {
+        fn name(&self) -> &'static str {
+            "SortRequiredExec"
+        }
+
         fn as_any(&self) -> &dyn std::any::Any {
             self
         }

datafusion/core/src/physical_optimizer/output_requirements.rs

@@ -133,6 +133,10 @@ impl DisplayAs for OutputRequirementExec {
 }
 
 impl ExecutionPlan for OutputRequirementExec {
+    fn name(&self) -> &'static str {
+        "OutputRequirementExec"
+    }
+
     fn as_any(&self) -> &dyn std::any::Any {
         self
     }

datafusion/core/src/physical_planner.rs

@@ -2625,6 +2625,10 @@ mod tests {
     }
 
     impl ExecutionPlan for NoOpExecutionPlan {
+        fn name(&self) -> &'static str {
+            "NoOpExecutionPlan"
+        }
+
         /// Return a reference to Any that can be used for downcasting
         fn as_any(&self) -> &dyn Any {
             self

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

@@ -636,6 +636,10 @@ impl DisplayAs for AggregateExec {
 }
 
 impl ExecutionPlan for AggregateExec {
+    fn name(&self) -> &'static str {
+        "AggregateExec"
+    }
+
     /// Return a reference to Any that can be used for down-casting
     fn as_any(&self) -> &dyn Any {
         self
@@ -1658,6 +1662,10 @@ mod tests {
     }
 
     impl ExecutionPlan for TestYieldingExec {
+        fn name(&self) -> &'static str {
+            "TestYieldingExec"
+        }
+
         fn as_any(&self) -> &dyn Any {
             self
         }

datafusion/physical-plan/src/analyze.rs

@@ -111,6 +111,10 @@ impl DisplayAs for AnalyzeExec {
 }
 
 impl ExecutionPlan for AnalyzeExec {
+    fn name(&self) -> &'static str {
+        "AnalyzeExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/coalesce_batches.rs

@@ -104,6 +104,10 @@ impl DisplayAs for CoalesceBatchesExec {
 }
 
 impl ExecutionPlan for CoalesceBatchesExec {
+    fn name(&self) -> &'static str {
+        "CoalesceBatchesExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -89,6 +89,10 @@ impl DisplayAs for CoalescePartitionsExec {
 }
 
 impl ExecutionPlan for CoalescePartitionsExec {
+    fn name(&self) -> &'static str {
+        "CoalescePartitionsExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/display.rs

@@ -489,6 +489,10 @@ mod tests {
     }
 
     impl ExecutionPlan for TestStatsExecPlan {
+        fn name(&self) -> &'static str {
+            "TestStatsExecPlan"
+        }
+
         fn as_any(&self) -> &dyn std::any::Any {
             self
         }

datafusion/physical-plan/src/empty.rs

@@ -101,6 +101,10 @@ impl DisplayAs for EmptyExec {
 }
 
 impl ExecutionPlan for EmptyExec {
+    fn name(&self) -> &'static str {
+        "EmptyExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/explain.rs

@@ -98,6 +98,10 @@ impl DisplayAs for ExplainExec {
 }
 
 impl ExecutionPlan for ExplainExec {
+    fn name(&self) -> &'static str {
+        "ExplainExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/filter.rs

@@ -231,6 +231,10 @@ impl DisplayAs for FilterExec {
 }
 
 impl ExecutionPlan for FilterExec {
+    fn name(&self) -> &'static str {
+        "FilterExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/insert.rs

@@ -206,6 +206,10 @@ impl DisplayAs for FileSinkExec {
 }
 
 impl ExecutionPlan for FileSinkExec {
+    fn name(&self) -> &'static str {
+        "FileSinkExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

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

@@ -194,6 +194,10 @@ impl DisplayAs for CrossJoinExec {
 }
 
 impl ExecutionPlan for CrossJoinExec {
+    fn name(&self) -> &'static str {
+        "CrossJoinExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -611,6 +611,10 @@ fn project_index_to_exprs(
 }
 
 impl ExecutionPlan for HashJoinExec {
+    fn name(&self) -> &'static str {
+        "HashJoinExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -205,6 +205,10 @@ impl DisplayAs for NestedLoopJoinExec {
 }
 
 impl ExecutionPlan for NestedLoopJoinExec {
+    fn name(&self) -> &'static str {
+        "NestedLoopJoinExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -262,6 +262,10 @@ impl DisplayAs for SortMergeJoinExec {
 }
 
 impl ExecutionPlan for SortMergeJoinExec {
+    fn name(&self) -> &'static str {
+        "SortMergeJoinExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -385,6 +385,10 @@ impl DisplayAs for SymmetricHashJoinExec {
 }
 
 impl ExecutionPlan for SymmetricHashJoinExec {
+    fn name(&self) -> &'static str {
+        "SymmetricHashJoinExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

datafusion/physical-plan/src/lib.rs

@@ -113,6 +113,15 @@ pub use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream};
 /// [`required_input_distribution`]: ExecutionPlan::required_input_distribution
 /// [`required_input_ordering`]: ExecutionPlan::required_input_ordering
 pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
+    /// Short name for the ExecutionPlan, such as 'ParquetExec'.
+    fn name(&self) -> &'static str {
+        let full_name = std::any::type_name::<Self>();
+        let maybe_start_idx = full_name.rfind(":");
+        match maybe_start_idx {
+            Some(start_idx) => &full_name[start_idx + 1..],
+            None => "UNKNOWN",
+        }
+    }
     /// Returns the execution plan as [`Any`] so that it can be
     /// downcast to a specific implementation.
     fn as_any(&self) -> &dyn Any;

datafusion/physical-plan/src/limit.rs

@@ -111,6 +111,10 @@ impl DisplayAs for GlobalLimitExec {
 }
 
 impl ExecutionPlan for GlobalLimitExec {
+    fn name(&self) -> &'static str {
+        "GlobalLimitExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self
@@ -317,6 +321,10 @@ impl DisplayAs for LocalLimitExec {
 }
 
 impl ExecutionPlan for LocalLimitExec {
+    fn name(&self) -> &'static str {
+        "LocalLimitExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/memory.rs

@@ -103,6 +103,10 @@ impl DisplayAs for MemoryExec {
 }
 
 impl ExecutionPlan for MemoryExec {
+    fn name(&self) -> &'static str {
+        "MemoryExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/placeholder_row.rs

@@ -119,6 +119,10 @@ impl DisplayAs for PlaceholderRowExec {
 }
 
 impl ExecutionPlan for PlaceholderRowExec {
+    fn name(&self) -> &'static str {
+        "PlaceholderRowExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/projection.rs

@@ -180,6 +180,10 @@ impl DisplayAs for ProjectionExec {
 }
 
 impl ExecutionPlan for ProjectionExec {
+    fn name(&self) -> &'static str {
+        "ProjectionExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

datafusion/physical-plan/src/recursive_query.rs

@@ -108,6 +108,10 @@ impl RecursiveQueryExec {
 }
 
 impl ExecutionPlan for RecursiveQueryExec {
+    fn name(&self) -> &'static str {
+        "RecursiveQueryExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -406,6 +406,10 @@ impl DisplayAs for RepartitionExec {
 }
 
 impl ExecutionPlan for RepartitionExec {
+    fn name(&self) -> &'static str {
+        "RepartitionExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self

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

@@ -226,6 +226,10 @@ impl DisplayAs for PartialSortExec {
 }
 
 impl ExecutionPlan for PartialSortExec {
+    fn name(&self) -> &'static str {
+        "PartialSortExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -860,6 +860,10 @@ impl DisplayAs for SortExec {
 }
 
 impl ExecutionPlan for SortExec {
+    fn name(&self) -> &'static str {
+        "SortExec"
+    }
+
     fn as_any(&self) -> &dyn Any {
         self
     }

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

@@ -144,6 +144,10 @@ impl DisplayAs for SortPreservingMergeExec {
 }
 
 impl ExecutionPlan for SortPreservingMergeExec {
+    fn name(&self) -> &'static str {
+        "SortPreservingMergeExec"
+    }
+
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
         self