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/03/26 06:01:36 UTC
[GitHub] [arrow] jacobBaumbach opened a new pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
jacobBaumbach opened a new pull request #9806:
URL: https://github.com/apache/arrow/pull/9806
Update output_partitioning for union to preserve Hash. Still needs a test for the hash case.
--
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
[GitHub] [arrow] jacobBaumbach commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812722697
I am hitting the following error in CI/CD
```
npm WARN tar ENOSPC: no space left on device, write
npm WARN tar ENOSPC: no space left on device, write
npm ERR! code ENOSPC
npm ERR! syscall write
npm ERR! errno -28
npm ERR! nospc ENOSPC: no space left on device, write
npm ERR! nospc There appears to be insufficient space on your system to finish.
npm ERR! nospc Clear up some disk space and try again.
```
As seen from the logs [here](https://github.com/apache/arrow/pull/9806/checks?check_run_id=2256351466#step:8:4512)
An issue has already been created for this [here](https://issues.apache.org/jira/projects/ARROW/issues/ARROW-12112?filter=allopenissues), which #9879 appears to fix.
Once #9879 is merged I am pretty confident ci/cd will pass.
--
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
[GitHub] [arrow] Dandandan commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807984850
> @Dandandan do you know of a good example test where the underlying `Partitioning` is of type `Hash`? I am having trouble finding one in the codebase, which is preventing me from creating a test for the `Hash` case. Thanks!
You are right, it is pretty new and unused. There is a lower level test for the `Hash` partitioning but not much else is using it yet.
In the DataFrame api you can pass `.repartition(Partitioning::Hash(expr))`
See here the relevant code/example for RoundRobinBatch:
https://github.com/apache/arrow/blob/a2a9f5d0b70d304817e65e32fbe1deef98ca7cb8/rust/datafusion/src/dataframe.rs#L190-L207
--
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
[GitHub] [arrow] codecov-io commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807968592
# [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=h1) Report
> Merging [#9806](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=desc) (1072d28) into [master](https://codecov.io/gh/apache/arrow/commit/29feea07fea118ad6ffc4d2615a5798412fe2f19?el=desc) (29feea0) will **decrease** coverage by `0.32%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #9806 +/- ##
==========================================
- Coverage 82.59% 82.26% -0.33%
==========================================
Files 248 252 +4
Lines 58294 58620 +326
==========================================
+ Hits 48149 48226 +77
- Misses 10145 10394 +249
```
| [Impacted Files](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [rust/arrow/src/array/transform/fixed\_binary.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL2ZpeGVkX2JpbmFyeS5ycw==) | `0.00% <0.00%> (-78.95%)` | :arrow_down: |
| [rust/arrow/src/array/transform/null.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL251bGwucnM=) | `0.00% <0.00%> (-66.67%)` | :arrow_down: |
| [rust/arrow/src/array/transform/structure.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3N0cnVjdHVyZS5ycw==) | `44.44% <0.00%> (-38.89%)` | :arrow_down: |
| [rust/arrow/src/array/transform/primitive.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3ByaW1pdGl2ZS5ycw==) | `71.42% <0.00%> (-28.58%)` | :arrow_down: |
| [rust/arrow/src/array/transform/mod.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL21vZC5ycw==) | `70.23% <0.00%> (-22.96%)` | :arrow_down: |
| [rust/arrow/src/array/transform/variable\_size.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3ZhcmlhYmxlX3NpemUucnM=) | `87.87% <0.00%> (-12.13%)` | :arrow_down: |
| [rust/arrow/src/array/array\_binary.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvYXJyYXlfYmluYXJ5LnJz) | `79.84% <0.00%> (-11.80%)` | :arrow_down: |
| [rust/arrow/src/buffer/mutable.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYnVmZmVyL211dGFibGUucnM=) | `83.89% <0.00%> (-6.87%)` | :arrow_down: |
| [rust/parquet/src/arrow/schema.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9wYXJxdWV0L3NyYy9hcnJvdy9zY2hlbWEucnM=) | `86.57% <0.00%> (-5.79%)` | :arrow_down: |
| [rust/arrow/src/array/array\_union.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvYXJyYXlfdW5pb24ucnM=) | `87.17% <0.00%> (-2.37%)` | :arrow_down: |
| ... and [58 more](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=footer). Last update [a2a9f5d...1072d28](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
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
[GitHub] [arrow] Dandandan commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606780118
##########
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:
So it doesn't say anything about this batch size, only he number of partitions.
The implementation of the Union creates the sum of input partitions as output, so each partition can be executed in parallel, so if we have 10 partitions in `a` and 20 in `b`, the `a UNION b` will have 30 partitions.
--
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
[GitHub] [arrow] Dandandan commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606779504
##########
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:
`RoundRobinBatch(1000)` means it still has 1000 partitions as output after repartitioning with round robin.
--
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
[GitHub] [arrow] Dandandan commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606779019
##########
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 think so too. Maybe just look for the first input (there is always at least 2 in union) and compare with the others to have same expression and same nr of partitions.
--
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
[GitHub] [arrow] jacobBaumbach commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807962699
@Dandandan do you know of a good example test where the underlying `Partition` is `Hash`? I am having trouble finding one in the codebase, which is preventing me from creating a test for the `Hash` case. Thanks!
--
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
[GitHub] [arrow] jacobBaumbach commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812696896
> Less valuable I think indeed, but for consistency/debugging/etc I think good to maintain both 👍
Awesome. I included it and tested that case. Once I get ci/cd to pass I will mark this as ready. Thank you for your help!
--
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
[GitHub] [arrow] alamb commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812177566
If nothing currently uses hash partitioning, perhaps we can remove it from the enum for now?
--
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
[GitHub] [arrow] Dandandan commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812693849
> @Dandandan Does it make sense to also maintain `RoundRobin`? Based on my understanding it doesn't seem like there is a value add, but like I mentioned above I am new to the project.
Less valuable I think indeed, but for consistency/debugging/etc I think good to maintain both 👍
--
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
[GitHub] [arrow] andygrove commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
andygrove commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812199849
Yes, partitioned hash joins are needed to make Ballista scale with large data sets. I hope to start work on that pretty soon. It would be easier with the donation PR merged.
--
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
[GitHub] [arrow] alamb commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-822358712
The Apache Arrow Rust community is moving the Rust implementation into its own dedicated github repositories [arrow-rs](https://github.com/apache/arrow-rs) and [arrow-datafusion](https://github.com/apache/arrow-datafusion). It is likely we will not merge this PR into this repository
Please see the [mailing-list](https://lists.apache.org/thread.html/rce7c61cb3f703367dc00984530016e3fcb828e5a8035655fbcccf862%40%3Cdev.arrow.apache.org%3E) thread for more details
We expect the process to take a few days and will follow up with a migration plan for the in-flight PRs.
--
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
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r607244510
##########
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:
That is my understanding
--
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
[GitHub] [arrow] jacobBaumbach commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r609280913
##########
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 still new to rust, so sorry for the beginner questions. Thank you for your patience.
Since `PhysicalExpr` is a trait and not a struct or enum I am unable to use `#[derive(PartialEq)]`. Additionally I am not able to do the following: `pub trait PhysicalExpr: Send + Sync + Display + Debug + PartialEq`. My guess for why this doesn't work is the same reason the `derive` isn't working, which is that `PhysicalExpr` isn't a struct or enum. Do you know how to require all objects that implement a trait also implement `PatialEq` or is there a different way?
--
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
[GitHub] [arrow] codecov-io edited a comment on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807968592
# [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=h1) Report
> Merging [#9806](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=desc) (22892ba) into [master](https://codecov.io/gh/apache/arrow/commit/2b87dfc5da67dc39d18ccc1d49ed5212b468496e?el=desc) (2b87dfc) will **increase** coverage by `0.01%`.
> The diff coverage is `97.10%`.
[](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #9806 +/- ##
==========================================
+ Coverage 82.68% 82.70% +0.01%
==========================================
Files 256 256
Lines 60105 60167 +62
==========================================
+ Hits 49698 49760 +62
Misses 10407 10407
```
| [Impacted Files](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [rust/datafusion/src/physical\_plan/union.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9kYXRhZnVzaW9uL3NyYy9waHlzaWNhbF9wbGFuL3VuaW9uLnJz) | `94.11% <97.10%> (+6.61%)` | :arrow_up: |
| [rust/parquet/src/encodings/encoding.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9wYXJxdWV0L3NyYy9lbmNvZGluZ3MvZW5jb2RpbmcucnM=) | `95.05% <0.00%> (+0.19%)` | :arrow_up: |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=footer). Last update [2b87dfc...22892ba](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
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
[GitHub] [arrow] jacobBaumbach commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r608322157
##########
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:
@alamb I am having trouble figuring out how to check for equality between `PhysicalExpr`. Do I need to implement a `PartialEq` or is there a more clever way?
--
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
[GitHub] [arrow] github-actions[bot] commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807960687
https://issues.apache.org/jira/browse/ARROW-11991
--
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
[GitHub] [arrow] jacobBaumbach commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r610329683
##########
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:
So I added `PartialEq` and ensured all structs that impl `PhysicalExpr` also `impl PartialEq`. For every occurrence of `Arc<dyn PhysicalExpr>` I get the following error, 132 times:
```bash
error[E0038]: the trait `PhysicalExpr` cannot be made into an object
--> datafusion/src/physical_plan/udf.rs:95:17
|
95 | args: &[Arc<dyn PhysicalExpr>],
| ^^^^^^^^^^^^^^^^ `PhysicalExpr` cannot be made into an object
|
note: for a trait to be "object safe" it needs to allow building a vtable to allow the call to be resolvable dynamically; for more information visit <https://doc.rust-lang.org/reference/items/traits.html#object-safety>
--> datafusion/src/physical_plan/mod.rs:188:57
|
188 | pub trait PhysicalExpr: Send + Sync + Display + Debug + PartialEq {
| ------------ ^^^^^^^^^ ...because it uses `Self` as a type parameter
| |
| this trait cannot be made into an object...
```
It appears that the inclusion of PartialEq makes it so `PhysicalExpr` doesn't play nicely with dynamic dispatch. Due to PartialEq taking a type arg of `Self`, which can't be determined under dynamic dispatch. I am not sure how to work around this.
--
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
[GitHub] [arrow] alamb commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-831243787
https://github.com/apache/arrow/pull/10096 has removed the arrow implementation from this repository (it now resides in https://github.com/apache/arrow-rs and https://github.com/apache/arrow-datafusion) in the hopes of streamlining the development process
Please re-target this PR (let us know if you need help doing so) to one/both of the new repositories.
Thank you for understanding and helping to make arrow-rs and datafusion better
--
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
[GitHub] [arrow] jacobBaumbach commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606825668
##########
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:
Just to make sure I am on the same page:
Two `Partitioning::Hash` may only be composed if they have the same expression and number of partitions. If the two `Partitioning::Hash` differ in expression and/or number of partitions then it becomes a `UnknownPartitioning`. Correct?
--
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
[GitHub] [arrow] jacobBaumbach commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812336401
@Dandandan Does it make sense to also maintain `RoundRobin`? Based on my understanding it doesn't seem like there is a value add, but like I mentioned above I am new to the project.
--
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
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r608882181
##########
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 suspect we will need to implement `PartialEq` (which you may be able to do via `#[derive(PartialEq)]` annotation on `PhysicalExpr`
--
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
[GitHub] [arrow] Dandandan commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812187846
> If nothing currently uses hash partitioning, perhaps we can remove it from the enum for now?
At least @andygrove was planning to add partitioned hash joins for use in Ballista, so the addition to be able to use it in combination with union would be useful?
--
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
[GitHub] [arrow] alamb closed pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb closed pull request #9806:
URL: https://github.com/apache/arrow/pull/9806
--
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
[GitHub] [arrow] jacobBaumbach commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r609280913
##########
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 still new to rust, so sorry for the beginner questions. Thank you for your patience.
Since `PhysicalExpr` is a trait and not a struct or enum I am unable to use `#[derive(PartialEq)]`. Additionally I am not able to do the following: `pub trait PhysicalExpr: Send + Sync + Display + Debug + PartialEq`. My guess is for why this doesn't work is the same reason the `derive` isn't working, which is that `PhysicalExpr` isn't a struct or enum. Do you know how to require all objects that implement a trait also implement `PatialEq` or is there a different way?
--
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
[GitHub] [arrow] jacobBaumbach commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812293918
Sorry for the slow progress. Will work on this tomorrow. I am new to arrow and realized my initial question was poorly worded. After reviewing the code I believe I can set a `Hash` partition by wrapping the `CSVExec` in a `RepartitionExec`, which I can then pass to `UnionExec`.
--
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
[GitHub] [arrow] jacobBaumbach edited a comment on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
jacobBaumbach edited a comment on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807962699
@Dandandan do you know of a good example test where the underlying `Partitioning` is of type `Hash`? I am having trouble finding one in the codebase, which is preventing me from creating a test for the `Hash` case. Thanks!
--
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
[GitHub] [arrow] Dandandan commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r606780118
##########
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:
So it doesn't say anything about this batch size, only he number of partitions.
The implementation of the Union creates the sum of input partitions as output, so each partition can be executed in parallel, so if we have 10 partitions in `a` and 20 in `b`, the `a UNION b` will have 30 partitions.
##########
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:
So it doesn't say anything about this batch size, only he number of partitions.
The implementation of the Union creates the sum of input partitions as output, so each partition can be executed in parallel, so if we have 10 partitions in `a` and 20 in `b`, `a UNION b` will have 30 partitions.
--
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
[GitHub] [arrow] alamb commented on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-812898925
@jacobBaumbach the no space left on device error is not caused by this PR
--
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
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r609651423
##########
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 would have expected
```
pub trait PhysicalExpr: Send + Sync + Display + Debug + PartialEq
```
To be the correct way as it says that everything that implements `PhysicalExpr` also needs to implement `PartialEq`. You would then have to `#[derive(PartialEq)]` or `impl PartialEq` for all structs that claim to `impl PhysicalExpr`
What error did you get when you tried that?
--
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
[GitHub] [arrow] alamb commented on a change in pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#discussion_r610912000
##########
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:
🤔 it seems [stackoverflow](https://stackoverflow.com/questions/25339603/how-to-test-for-equality-between-trait-objects) has some context
Perhap we need to implement our own version of equality for `PhysicalExpr` or something.
I am not sure how to proceed, to be honest. @Dandandan any thoughts?
--
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
[GitHub] [arrow] codecov-io edited a comment on pull request #9806: ARROW-11991: [Rust][DataFusion] Maintain partition information in Union
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #9806:
URL: https://github.com/apache/arrow/pull/9806#issuecomment-807968592
# [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=h1) Report
> Merging [#9806](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=desc) (2fbdc80) into [master](https://codecov.io/gh/apache/arrow/commit/29feea07fea118ad6ffc4d2615a5798412fe2f19?el=desc) (29feea0) will **increase** coverage by `0.07%`.
> The diff coverage is `n/a`.
[](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #9806 +/- ##
==========================================
+ Coverage 82.59% 82.66% +0.07%
==========================================
Files 248 256 +8
Lines 58294 60035 +1741
==========================================
+ Hits 48149 49630 +1481
- Misses 10145 10405 +260
```
| [Impacted Files](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [rust/arrow/src/array/transform/fixed\_binary.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL2ZpeGVkX2JpbmFyeS5ycw==) | `5.26% <0.00%> (-73.69%)` | :arrow_down: |
| [rust/arrow/src/array/transform/null.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL251bGwucnM=) | `0.00% <0.00%> (-66.67%)` | :arrow_down: |
| [rust/arrow/src/array/transform/structure.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3N0cnVjdHVyZS5ycw==) | `44.44% <0.00%> (-38.89%)` | :arrow_down: |
| [rust/arrow/src/array/transform/primitive.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3ByaW1pdGl2ZS5ycw==) | `71.42% <0.00%> (-28.58%)` | :arrow_down: |
| [rust/arrow/src/array/transform/mod.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL21vZC5ycw==) | `70.23% <0.00%> (-22.96%)` | :arrow_down: |
| [rust/arrow/src/array/transform/variable\_size.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvdHJhbnNmb3JtL3ZhcmlhYmxlX3NpemUucnM=) | `87.87% <0.00%> (-12.13%)` | :arrow_down: |
| [rust/arrow/src/array/array\_binary.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9hcnJvdy9zcmMvYXJyYXkvYXJyYXlfYmluYXJ5LnJz) | `86.05% <0.00%> (-5.58%)` | :arrow_down: |
| [rust/parquet/src/schema/parser.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9wYXJxdWV0L3NyYy9zY2hlbWEvcGFyc2VyLnJz) | `86.27% <0.00%> (-3.93%)` | :arrow_down: |
| [rust/datafusion/src/physical\_plan/group\_scalar.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9kYXRhZnVzaW9uL3NyYy9waHlzaWNhbF9wbGFuL2dyb3VwX3NjYWxhci5ycw==) | `61.84% <0.00%> (-3.44%)` | :arrow_down: |
| [rust/parquet/src/arrow/schema.rs](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree#diff-cnVzdC9wYXJxdWV0L3NyYy9hcnJvdy9zY2hlbWEucnM=) | `89.08% <0.00%> (-3.29%)` | :arrow_down: |
| ... and [79 more](https://codecov.io/gh/apache/arrow/pull/9806/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=footer). Last update [a2a9f5d...2fbdc80](https://codecov.io/gh/apache/arrow/pull/9806?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
--
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