You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/04/04 09:59:05 UTC
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
alamb commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606777115
##########
File path: rust/datafusion/src/physical_plan/union.rs
##########
@@ -60,15 +60,31 @@ impl ExecutionPlan for UnionExec {
/// Output of the union is the combination of all output partitions of the inputs
fn output_partitioning(&self) -> Partitioning {
- // Sums all the output partitions
- let num_partitions = self
- .inputs
+ let intial: Option<Partitioning> = None;
+ self.inputs
.iter()
- .map(|plan| plan.output_partitioning().partition_count())
- .sum();
- // TODO: this loses partitioning info in case of same partitioning scheme (for example `Partitioning::Hash`)
- // https://issues.apache.org/jira/browse/ARROW-11991
- Partitioning::UnknownPartitioning(num_partitions)
+ .fold(intial, |acc, input| {
+ match (acc, input.output_partitioning()) {
+ (None, partition) => Some(partition),
+ (
+ Some(Partitioning::Hash(mut vector_acc, size_acc)),
+ Partitioning::Hash(vector, size),
+ ) => {
+ vector_acc.append(&mut vector.clone());
Review comment:
I am not sure this is correct -- of the input is hash partitioned by some expression across `size_acc` partitions, the output will only be hash partitioned if the hash expressions are the same and the number of output partitions is the same, right?
##########
File path: rust/datafusion/src/physical_plan/union.rs
##########
@@ -60,15 +60,31 @@ impl ExecutionPlan for UnionExec {
/// Output of the union is the combination of all output partitions of the inputs
fn output_partitioning(&self) -> Partitioning {
- // Sums all the output partitions
- let num_partitions = self
- .inputs
+ let intial: Option<Partitioning> = None;
+ self.inputs
.iter()
- .map(|plan| plan.output_partitioning().partition_count())
- .sum();
- // TODO: this loses partitioning info in case of same partitioning scheme (for example `Partitioning::Hash`)
- // https://issues.apache.org/jira/browse/ARROW-11991
- Partitioning::UnknownPartitioning(num_partitions)
+ .fold(intial, |acc, input| {
+ match (acc, input.output_partitioning()) {
+ (None, partition) => Some(partition),
+ (
+ Some(Partitioning::Hash(mut vector_acc, size_acc)),
+ Partitioning::Hash(vector, size),
+ ) => {
+ vector_acc.append(&mut vector.clone());
+ Some(Partitioning::Hash(vector_acc, size_acc + size))
+ }
+ (
+ Some(Partitioning::RoundRobinBatch(size_acc)),
+ Partitioning::RoundRobinBatch(size),
+ ) => Some(Partitioning::RoundRobinBatch(size_acc + size)),
Review comment:
I think the output batch size might be more precisely described as `max(size_acc, size)`
though I am unsure of the requirements on a operator that produces `RoundRobinBatch(1000)` -- does that mean all batches must be 1000? The docs on [`Partitioning`](https://github.com/apache/arrow/blob/8e43f23dcc6a9e630516228f110c48b64d13cec6/rust/datafusion/src/physical_plan/mod.rs#L131) are not super clear
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org