Skip to content

Commit 8e9a398

Browse files
qrilkaqrilka
committed
Set DisplayAs to be a supertrait of ExecutionPlan
Follow-up to apache#6711
1 parent 59d1b81 commit 8e9a398

40 files changed

Lines changed: 880 additions & 746 deletions

‎datafusion-examples/examples/custom_datasource.rs‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,13 +27,14 @@ use datafusion::execution::context::{SessionState, TaskContext};
2727
use datafusion::physical_plan::expressions::PhysicalSortExpr;
2828
use datafusion::physical_plan::memory::MemoryStream;
2929
use datafusion::physical_plan::{
30-
project_schema, ExecutionPlan, SendableRecordBatchStream, Statistics,
30+
project_schema, DisplayAs, DisplayFormatType, ExecutionPlan,
31+
SendableRecordBatchStream, Statistics,
3132
};
3233
use datafusion::prelude::*;
3334
use datafusion_expr::{Expr, LogicalPlanBuilder};
3435
use std::any::Any;
3536
use std::collections::{BTreeMap, HashMap};
36-
use std::fmt::{Debug, Formatter};
37+
use std::fmt::{self, Debug, Formatter};
3738
use std::sync::{Arc, Mutex};
3839
use std::time::Duration;
3940
use tokio::time::timeout;
@@ -204,6 +205,12 @@ impl CustomExec {
204205
}
205206
}
206207

208+
impl DisplayAs for CustomExec {
209+
fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter) -> std::fmt::Result {
210+
write!(f, "CustomExec")
211+
}
212+
}
213+
207214
impl ExecutionPlan for CustomExec {
208215
fn as_any(&self) -> &dyn Any {
209216
self

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

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,17 @@ impl ArrowExec {
6868
}
6969
}
7070

71+
impl DisplayAs for ArrowExec {
72+
fn fmt_as(
73+
&self,
74+
t: DisplayFormatType,
75+
f: &mut std::fmt::Formatter,
76+
) -> std::fmt::Result {
77+
write!(f, "ArrowExec: ")?;
78+
self.base_config.fmt_as(t, f)
79+
}
80+
}
81+
7182
impl ExecutionPlan for ArrowExec {
7283
fn as_any(&self) -> &dyn Any {
7384
self
@@ -132,15 +143,6 @@ impl ExecutionPlan for ArrowExec {
132143
Some(self.metrics.clone_inner())
133144
}
134145

135-
fn fmt_as(
136-
&self,
137-
t: DisplayFormatType,
138-
f: &mut std::fmt::Formatter,
139-
) -> std::fmt::Result {
140-
write!(f, "ArrowExec: ")?;
141-
self.base_config.fmt_as(t, f)
142-
}
143-
144146
fn statistics(&self) -> Statistics {
145147
self.projected_statistics.clone()
146148
}

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

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,17 @@ impl AvroExec {
6565
}
6666
}
6767

68+
impl DisplayAs for AvroExec {
69+
fn fmt_as(
70+
&self,
71+
t: DisplayFormatType,
72+
f: &mut std::fmt::Formatter,
73+
) -> std::fmt::Result {
74+
write!(f, "AvroExec: ")?;
75+
self.base_config.fmt_as(t, f)
76+
}
77+
}
78+
6879
impl ExecutionPlan for AvroExec {
6980
fn as_any(&self) -> &dyn Any {
7081
self
@@ -141,15 +152,6 @@ impl ExecutionPlan for AvroExec {
141152
Ok(Box::pin(stream))
142153
}
143154

144-
fn fmt_as(
145-
&self,
146-
t: DisplayFormatType,
147-
f: &mut std::fmt::Formatter,
148-
) -> std::fmt::Result {
149-
write!(f, "AvroExec: ")?;
150-
self.base_config.fmt_as(t, f)
151-
}
152-
153155
fn statistics(&self) -> Statistics {
154156
self.projected_statistics.clone()
155157
}

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

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,18 @@ impl CsvExec {
9999
}
100100
}
101101

102+
impl DisplayAs for CsvExec {
103+
fn fmt_as(
104+
&self,
105+
t: DisplayFormatType,
106+
f: &mut std::fmt::Formatter,
107+
) -> std::fmt::Result {
108+
write!(f, "CsvExec: ")?;
109+
self.base_config.fmt_as(t, f)?;
110+
write!(f, ", has_header={}", self.has_header)
111+
}
112+
}
113+
102114
impl ExecutionPlan for CsvExec {
103115
/// Return a reference to Any that can be used for downcasting
104116
fn as_any(&self) -> &dyn Any {
@@ -172,16 +184,6 @@ impl ExecutionPlan for CsvExec {
172184
Ok(Box::pin(stream) as SendableRecordBatchStream)
173185
}
174186

175-
fn fmt_as(
176-
&self,
177-
t: DisplayFormatType,
178-
f: &mut std::fmt::Formatter,
179-
) -> std::fmt::Result {
180-
write!(f, "CsvExec: ")?;
181-
self.base_config.fmt_as(t, f)?;
182-
write!(f, ", has_header={}", self.has_header)
183-
}
184-
185187
fn statistics(&self) -> Statistics {
186188
self.projected_statistics.clone()
187189
}

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

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,17 @@ impl NdJsonExec {
8585
}
8686
}
8787

88+
impl DisplayAs for NdJsonExec {
89+
fn fmt_as(
90+
&self,
91+
t: DisplayFormatType,
92+
f: &mut std::fmt::Formatter,
93+
) -> std::fmt::Result {
94+
write!(f, "JsonExec: ")?;
95+
self.base_config.fmt_as(t, f)
96+
}
97+
}
98+
8899
impl ExecutionPlan for NdJsonExec {
89100
fn as_any(&self) -> &dyn Any {
90101
self
@@ -150,15 +161,6 @@ impl ExecutionPlan for NdJsonExec {
150161
Ok(Box::pin(stream) as SendableRecordBatchStream)
151162
}
152163

153-
fn fmt_as(
154-
&self,
155-
t: DisplayFormatType,
156-
f: &mut std::fmt::Formatter,
157-
) -> std::fmt::Result {
158-
write!(f, "JsonExec: ")?;
159-
self.base_config.fmt_as(t, f)
160-
}
161-
162164
fn statistics(&self) -> Statistics {
163165
self.projected_statistics.clone()
164166
}

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

Lines changed: 28 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -325,6 +325,34 @@ impl ParquetExec {
325325
}
326326
}
327327

328+
impl DisplayAs for ParquetExec {
329+
fn fmt_as(
330+
&self,
331+
t: DisplayFormatType,
332+
f: &mut std::fmt::Formatter,
333+
) -> std::fmt::Result {
334+
match t {
335+
DisplayFormatType::Default | DisplayFormatType::Verbose => {
336+
let predicate_string = self
337+
.predicate
338+
.as_ref()
339+
.map(|p| format!(", predicate={p}"))
340+
.unwrap_or_default();
341+
342+
let pruning_predicate_string = self
343+
.pruning_predicate
344+
.as_ref()
345+
.map(|pre| format!(", pruning_predicate={}", pre.predicate_expr()))
346+
.unwrap_or_default();
347+
348+
write!(f, "ParquetExec: ")?;
349+
self.base_config.fmt_as(t, f)?;
350+
write!(f, "{}{}", predicate_string, pruning_predicate_string,)
351+
}
352+
}
353+
}
354+
}
355+
328356
impl ExecutionPlan for ParquetExec {
329357
/// Return a reference to Any that can be used for downcasting
330358
fn as_any(&self) -> &dyn Any {
@@ -413,32 +441,6 @@ impl ExecutionPlan for ParquetExec {
413441
Ok(Box::pin(stream))
414442
}
415443

416-
fn fmt_as(
417-
&self,
418-
t: DisplayFormatType,
419-
f: &mut std::fmt::Formatter,
420-
) -> std::fmt::Result {
421-
match t {
422-
DisplayFormatType::Default | DisplayFormatType::Verbose => {
423-
let predicate_string = self
424-
.predicate
425-
.as_ref()
426-
.map(|p| format!(", predicate={p}"))
427-
.unwrap_or_default();
428-
429-
let pruning_predicate_string = self
430-
.pruning_predicate
431-
.as_ref()
432-
.map(|pre| format!(", pruning_predicate={}", pre.predicate_expr()))
433-
.unwrap_or_default();
434-
435-
write!(f, "ParquetExec: ")?;
436-
self.base_config.fmt_as(t, f)?;
437-
write!(f, "{}{}", predicate_string, pruning_predicate_string,)
438-
}
439-
}
440-
}
441-
442444
fn metrics(&self) -> Option<MetricsSet> {
443445
Some(self.metrics.clone_inner())
444446
}

‎datafusion/core/src/physical_optimizer/repartition.rs‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -338,7 +338,7 @@ mod tests {
338338
use crate::physical_plan::sorts::sort::SortExec;
339339
use crate::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
340340
use crate::physical_plan::union::UnionExec;
341-
use crate::physical_plan::{displayable, DisplayFormatType, Statistics};
341+
use crate::physical_plan::{displayable, DisplayAs, DisplayFormatType, Statistics};
342342
use datafusion_physical_expr::PhysicalSortRequirement;
343343

344344
fn schema() -> SchemaRef {
@@ -1136,6 +1136,16 @@ mod tests {
11361136
}
11371137
}
11381138

1139+
impl DisplayAs for SortRequiredExec {
1140+
fn fmt_as(
1141+
&self,
1142+
_t: DisplayFormatType,
1143+
f: &mut std::fmt::Formatter,
1144+
) -> std::fmt::Result {
1145+
write!(f, "SortRequiredExec")
1146+
}
1147+
}
1148+
11391149
impl ExecutionPlan for SortRequiredExec {
11401150
fn as_any(&self) -> &dyn std::any::Any {
11411151
self
@@ -1184,13 +1194,5 @@ mod tests {
11841194
fn statistics(&self) -> Statistics {
11851195
self.input.statistics()
11861196
}
1187-
1188-
fn fmt_as(
1189-
&self,
1190-
_t: DisplayFormatType,
1191-
f: &mut std::fmt::Formatter,
1192-
) -> std::fmt::Result {
1193-
write!(f, "SortRequiredExec")
1194-
}
11951197
}
11961198
}

0 commit comments

Comments
 (0)