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:
[email protected]
