Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 37 additions & 37 deletions ballista/client/tests/context_checks.rs
Original file line number Diff line number Diff line change
Expand Up @@ -998,43 +998,43 @@ mod supported {
.unwrap();

let expected: Vec<&str> = vec![
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
"| plan_type | plan |",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
"| logical_plan | Projection: count(Int64(1)) AS count(*), id |",
"| | Aggregate: groupBy=[[id]], aggr=[[count(Int64(1))]] |",
"| | Projection: __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1) AS UNNEST(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5))) AS id |",
"| | Unnest: lists[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))|depth=1] structs[] |",
"| | Projection: List([1, 2, 3, 4, 5]) AS __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5))) |",
"| | EmptyRelation: rows=1 |",
"| physical_plan | ProjectionExec: expr=[count(Int64(1))@1 as count(*), id@0 as id] |",
"| | AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | CoalesceBatchesExec: target_batch_size=8192 |",
"| | RepartitionExec: partitioning=Hash([id@0], 16), input_partitions=1 |",
"| | AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | ProjectionExec: expr=[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1)@0 as id] |",
"| | UnnestExec |",
"| | ProjectionExec: expr=[[1, 2, 3, 4, 5] as __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))] |",
"| | PlaceholderRowExec |",
"| | |",
"| distributed_plan | =========ResolvedStage[stage_id=1.0, partitions=1]========= |",
"| | ShuffleWriterExec: partitioning:Some(Hash([Column { name: \"id\", index: 0 }], 16)) |",
"| | AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | ProjectionExec: expr=[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1)@0 as id] |",
"| | UnnestExec |",
"| | ProjectionExec: expr=[[1, 2, 3, 4, 5] as __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))] |",
"| | PlaceholderRowExec |",
"| | |",
"| | =========UnResolvedStage[stage_id=2.0, children=1]========= |",
"| | Inputs{1: StageOutput { partition_locations: {}, complete: false }} |",
"| | ShuffleWriterExec: partitioning:None |",
"| | ProjectionExec: expr=[count(Int64(1))@1 as count(*), id@0 as id] |",
"| | AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | CoalesceBatchesExec: target_batch_size=8192 |",
"| | UnresolvedShuffleExec: partitioning=Hash([Column { name: \"id\", index: 0 }], 16) |",
"| | |",
"| | |",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
"| plan_type | plan |",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
"| logical_plan | Projection: count(Int64(1)) AS count(*), id |",
"| | Aggregate: groupBy=[[id]], aggr=[[count(Int64(1))]] |",
"| | Projection: __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1) AS UNNEST(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5))) AS id |",
"| | Unnest: lists[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))|depth=1] structs[] |",
"| | Projection: List([1, 2, 3, 4, 5]) AS __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5))) |",
"| | EmptyRelation: rows=1 |",
"| physical_plan | ProjectionExec: expr=[count(Int64(1))@1 as count(*), id@0 as id] |",
"| | AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | CoalesceBatchesExec: target_batch_size=8192 |",
"| | RepartitionExec: partitioning=Hash([id@0], 16), input_partitions=1 |",
"| | AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | ProjectionExec: expr=[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1)@0 as id] |",
"| | UnnestExec |",
"| | ProjectionExec: expr=[[1, 2, 3, 4, 5] as __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))] |",
"| | PlaceholderRowExec |",
"| | |",
"| distributed_plan | =========ResolvedStage[stage_id=1.0, partitions=1]========= |",
"| | ShuffleWriterExec: partitioning: Hash([id@0], 16) |",
"| | AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | ProjectionExec: expr=[__unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)),depth=1)@0 as id] |",
"| | UnnestExec |",
"| | ProjectionExec: expr=[[1, 2, 3, 4, 5] as __unnest_placeholder(make_array(Int64(1),Int64(2),Int64(3),Int64(4),Int64(5)))] |",
"| | PlaceholderRowExec |",
"| | |",
"| | =========UnResolvedStage[stage_id=2.0, children=1]========= |",
"| | Inputs{1: StageOutput { partition_locations: {}, complete: false }} |",
"| | ShuffleWriterExec: partitioning: None |",
"| | ProjectionExec: expr=[count(Int64(1))@1 as count(*), id@0 as id] |",
"| | AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[count(Int64(1))] |",
"| | CoalesceBatchesExec: target_batch_size=8192 |",
"| | UnresolvedShuffleExec: partitioning: Hash([id@0], 16) |",
"| | |",
"| | |",
"+------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+",
];
assert_batches_eq!(expected, &result);
}
Expand Down
4 changes: 2 additions & 2 deletions ballista/core/src/execution_plans/shuffle_reader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,12 +103,12 @@ impl DisplayAs for ShuffleReaderExec {
DisplayFormatType::Default | DisplayFormatType::Verbose => {
write!(
f,
"ShuffleReaderExec: partitioning={:?}",
"ShuffleReaderExec: partitioning: {}",
self.properties.partitioning,
)
}
DisplayFormatType::TreeRender => {
write!(f, "partitioning={:?}", self.properties.partitioning)
write!(f, "partitioning={}", self.properties.partitioning)
}
}
}
Expand Down
14 changes: 12 additions & 2 deletions ballista/core/src/execution_plans/shuffle_writer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -359,12 +359,22 @@ impl DisplayAs for ShuffleWriterExec {
DisplayFormatType::Default | DisplayFormatType::Verbose => {
write!(
f,
"ShuffleWriterExec: partitioning:{:?}",
"ShuffleWriterExec: partitioning: {}",
self.shuffle_output_partitioning
.as_ref()
.map(|p| p.to_string())
.unwrap_or("None".to_string())
)
}
DisplayFormatType::TreeRender => {
write!(f, "partitioning={:?}", self.shuffle_output_partitioning)
write!(
f,
"partitioning={}",
self.shuffle_output_partitioning
.as_ref()
.map(|p| p.to_string())
.unwrap_or("None".to_string())
)
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions ballista/core/src/execution_plans/unresolved_shuffle.rs
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,14 @@ impl DisplayAs for UnresolvedShuffleExec {
DisplayFormatType::Default | DisplayFormatType::Verbose => {
write!(
f,
"UnresolvedShuffleExec: partitioning={:?}",
"UnresolvedShuffleExec: partitioning: {}",
self.properties().output_partitioning()
)
}
DisplayFormatType::TreeRender => {
write!(
f,
"partitioning={:?}",
"partitioning={}",
self.properties().output_partitioning()
)
}
Expand Down