You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "mustafasrepo (via GitHub)" <gi...@apache.org> on 2023/07/28 15:36:14 UTC
[GitHub] [arrow-datafusion] mustafasrepo opened a new pull request, #7129: [bug]: Fix multi partition wrong column requirement bug
mustafasrepo opened a new pull request, #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes [#7128](https://github.com/apache/arrow-datafusion/issues/7128).
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
# What changes are included in this PR?
As explained in the [issue](https://github.com/apache/arrow-datafusion/issues/7128) body. The reason for this bug is that, during `Aggregate::Final` and `Aggregate::FinalPartitioned` stages, requirement for the order sensitive aggregators are no longer valid at the input schema (their input is `Aggregate::Partial`). Hence when these expressions are required by executor, they may refer to invalid or wrong columns.
This PR fixes this bug, by removing requirement for `Aggregate::FinalPartitioned` and `Aggregate::Final` modes. These modes receive complete data from different partitions. Order sensitive accumulators in their `merge_batch` method, accumulates data from different partitions by considering ordering of each partition. Hence we do not need any outside mechanism (such as adding a Sort to the plan) for correct operation.
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
# Are these changes tested?
Yes new tests are added
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277721280
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2946,6 +2956,39 @@ FRA [200.0, 50.0] 50 50
GRC [80.0, 30.0] 30 30
TUR [100.0, 75.0] 75 75
+# make sure that query below runs in multi partitions
+statement ok
+set datafusion.execution.target_partitions = 8;
+
+query ?
+SELECT ARRAY_AGG(e.rate ORDER BY e.sn)
Review Comment:
This is the buggy query in the issue
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1279180964
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2689,13 +2703,12 @@ SortPreservingMergeExec: [country@0 ASC NULLS LAST]
--SortExec: expr=[country@0 ASC NULLS LAST]
----ProjectionExec: expr=[country@0 as country, FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as fv1, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@2 as fv2]
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
---------SortExec: expr=[ts@1 ASC NULLS LAST]
-----------CoalesceBatchesExec: target_batch_size=8192
-------------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
---------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)], ordering_mode=None
-----------------SortExec: expr=[ts@1 ASC NULLS LAST]
-------------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
---------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------CoalesceBatchesExec: target_batch_size=8192
+----------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
+------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
Review Comment:
I am not sure this new plan is correct either as the data is not necessarily sorted for the final grouping . The old plan also seems wrong.
My reasoning is that for an aggregate like this:
```
FIRST_VALUE(amount ORDER BY ts ASC)
```
the input to the group operator has to be sorted on `ts`.
However, I don't see how the order by `ts` is preserved after the `AggregateExec: mode=Partial` first grouping phase
```shell
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
** I think the order by `ts` is not preserved here, so the data is not ordered by ts for the final grouping **
------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
```
Maybe the right solution would be to do a single phase grouping when any of the aggregates have an `ORDER BY` clause
```
AggregateExec: mode=Final, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)] ** no partial group by
SortExec: expr=[ts@1 ASC NULLS LAST]
RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
MemoryExec: partitions=1, partition_sizes=[1]
```
🤔
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2689,13 +2703,12 @@ SortPreservingMergeExec: [country@0 ASC NULLS LAST]
--SortExec: expr=[country@0 ASC NULLS LAST]
----ProjectionExec: expr=[country@0 as country, FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as fv1, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@2 as fv2]
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
---------SortExec: expr=[ts@1 ASC NULLS LAST]
-----------CoalesceBatchesExec: target_batch_size=8192
-------------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
---------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)], ordering_mode=None
-----------------SortExec: expr=[ts@1 ASC NULLS LAST]
-------------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
---------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------CoalesceBatchesExec: target_batch_size=8192
+----------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
+------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
Review Comment:
I am not sure this new plan is correct either as the data is not necessarily sorted for the final grouping . The old plan also seems wrong.
My reasoning is that for an aggregate like this:
```
FIRST_VALUE(amount ORDER BY ts ASC)
```
the input to the group operator has to be sorted on `ts`.
However, I don't see how the order by `ts` is preserved after the `AggregateExec: mode=Partial` first grouping phase
```shell
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
** I think the order by `ts` is not preserved here, so the data is not ordered by ts for the final grouping **
------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
```
Maybe the right solution would be to do a single phase grouping when any of the aggregates have an `ORDER BY` clause
```
AggregateExec: mode=Final, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)] ** no partial group by
SortExec: expr=[ts@1 ASC NULLS LAST]
RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
MemoryExec: partitions=1, partition_sizes=[1]
```
🤔
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1280437195
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1370,6 +1372,54 @@ mod tests {
)
}
+ /// some mock data to aggregates
+ fn some_data_v2() -> (Arc<Schema>, Vec<RecordBatch>) {
+ // define a schema.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::UInt32, false),
+ Field::new("b", DataType::Float64, false),
+ ]));
+
+ // define data.
Review Comment:
```suggestion
// Data is constructed such that first and last value result is at 2nd and 3rd partition.
// With this construction, we guarantee that expected result received is not accident,
// but merging works properly (It doesn't depend on the data insertion order).
```
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -1960,21 +1960,20 @@ SortPreservingMergeExec: [col0@0 ASC NULLS LAST]
--SortExec: expr=[col0@0 ASC NULLS LAST]
----ProjectionExec: expr=[col0@0 as col0, LAST_VALUE(r.col1) ORDER BY [r.col0 ASC NULLS LAST]@3 as last_col1]
------AggregateExec: mode=FinalPartitioned, gby=[col0@0 as col0, col1@1 as col1, col2@2 as col2], aggr=[LAST_VALUE(r.col1)]
---------SortExec: expr=[col0@3 ASC NULLS LAST]
Review Comment:
I it makes sense that there is nothing inserted between the two aggregate phases given the aggregates themselves track the sortedness
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277718979
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -1960,21 +1960,20 @@ SortPreservingMergeExec: [col0@0 ASC NULLS LAST]
--SortExec: expr=[col0@0 ASC NULLS LAST]
----ProjectionExec: expr=[col0@0 as col0, LAST_VALUE(r.col1) ORDER BY [r.col0 ASC NULLS LAST]@3 as last_col1]
------AggregateExec: mode=FinalPartitioned, gby=[col0@0 as col0, col1@1 as col1, col2@2 as col2], aggr=[LAST_VALUE(r.col1)]
---------SortExec: expr=[col0@3 ASC NULLS LAST]
Review Comment:
Generally in the plan `SortExec` before `AggregateExec: mode=FinalPartitioned` or `AggregateExec: mode=Final` is removed.
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1279106492
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
That makes sense. Sorry for my confusion
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278576304
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I should have stressed it, this change doesn't affect code path for GroupOrderingNone. I recognized that we already store aggregation as option as in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/mod.rs#L269). When this mode is `Option::None`, it is equivalent to `GroupByOrderMode::None`. So I thought, there is no need to keep track of `AggregationMode::None` separately. As can be seen in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/row_hash.rs#L239). When option is `None`, `GroupOrdering::None ` is used. If this is misleading, I can retract this change. It is not important, for 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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278576304
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I should have stressed it, but I recognized that we already store aggregation as option as in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/mod.rs#L269). When this mode is `Option::None`, it is equivalent to `GroupByOrderMode::None`. So I thought, there is no need to keep track of `AggregationMode::None` separately. As can be seen in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/row_hash.rs#L239). When option is `None`, `GroupOrdering::None ` is used. If this is misleading, I can retract this change. It is not important, for 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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278576304
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I should have stressed it, but I recognized that we already store aggregation as option as in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/mod.rs#L269). When this mode is `None`, it is equivalent to `GroupByOrderMode::None`. So I thought, there is no need to keep track represent this as mode separately. As can be seen in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/row_hash.rs#L239). When option is `None`, `GroupOrdering::None ` is used. If this is misleading, I can retract this change. It is not important, for 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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277722494
##########
datafusion/physical-expr/src/aggregate/array_agg_ordered.rs:
##########
@@ -228,11 +231,13 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator {
.iter()
.map(|sort_expr| sort_expr.options)
.collect::<Vec<_>>();
- self.values = merge_ordered_arrays(
+ let (new_values, new_orderings) = merge_ordered_arrays(
Review Comment:
Since we no longer rely on outside mechanism for merging, During merging we need to merge orderings (additional to values).
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1279277814
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2689,13 +2703,12 @@ SortPreservingMergeExec: [country@0 ASC NULLS LAST]
--SortExec: expr=[country@0 ASC NULLS LAST]
----ProjectionExec: expr=[country@0 as country, FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as fv1, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@2 as fv2]
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
---------SortExec: expr=[ts@1 ASC NULLS LAST]
-----------CoalesceBatchesExec: target_batch_size=8192
-------------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
---------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)], ordering_mode=None
-----------------SortExec: expr=[ts@1 ASC NULLS LAST]
-------------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
---------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------CoalesceBatchesExec: target_batch_size=8192
+----------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
+------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
Review Comment:
Consider table below
| amount | ts |
| -------- | ------- |
| 12 | 1|
| 11| 2|
| 13| 3|
| 12|4|
| 11| 5|
| 13| 6|
| 12 |7|
| 11 | 8|
| 13| 9|
Also assume we have 3 partitions, receiving following data
| amount | ts |
| -------- | ------- |
| 11| 2|
| 11| 5|
| 11 | 8|
| amount | ts |
| -------- | ------- |
| 12 | 1|
| 12 |4|
| 12 |7|
| amount | ts |
| -------- | ------- |
| 13| 3|
| 13| 6|
| 13| 9|
AggregatePartial would produce following values
(11, 2); (12, 1); (13;3) for each partition. First value represents first_value for this partitition. Second value represents its corresponding ts value. In this case
AgregateFinal would receive following batch
| amount | ts of amount partial result |
| -------- | ------- |
| 11| 2|
| 12| 1|
| 13| 3|.
During `merge_batch` method of `first_value` first value is calculated by considering `ts` values corresponding to `amount` for each partition. In our case, since requirement is `ts ASC`, first value should be from the row that have smallest `ts` (in our case 1). Hence result will be 12. Please note that `ts` at the final input and `ts` at the partial input doesn't correspond to same column. `ts` at the final aggregation input, comes from the `state` of aggregation partial result.
In short, we delegated responsibility to sort to `merge_batch` algorithm. Because, the column where sorting will be done is no longer valid for the aggregation final.
> Maybe the right solution would be to do a single phase grouping when any of the aggregates have an ORDER BY clause
this would certainly work. However, I wanted to use existing parallelization as much as possible. Hence, I wanted to make aggregators to work in Partial and Final modes.
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1279209471
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I got inspired / wanted some excuse to work on the code, so I made https://github.com/apache/arrow-datafusion/pull/7150 as a potential alternate approach
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277715398
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1885,6 +1933,134 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn run_first_last_multi_partitions() -> Result<()> {
+ for use_coalesce_batches in [false, true] {
+ for is_first_acc in [false, true] {
+ first_last_multi_partitions(use_coalesce_batches, is_first_acc).await?
+ }
+ }
+ Ok(())
+ }
+
+ // This function construct either physical plan below,
+ //
+ // "AggregateExec: mode=Final, gby=[a@0 as a], aggr=[FIRST_VALUE(b)]",
+ // " CoalesceBatchesExec: target_batch_size=1024",
+ // " CoalescePartitionsExec",
+ // " AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[FIRST_VALUE(b)], ordering_mode=None",
+ // " MemoryExec: partitions=4, partition_sizes=[1, 1, 1, 1]",
+ //
+ // or
+ //
+ // "AggregateExec: mode=Final, gby=[a@0 as a], aggr=[FIRST_VALUE(b)]",
+ // " CoalescePartitionsExec",
+ // " AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[FIRST_VALUE(b)], ordering_mode=None",
+ // " MemoryExec: partitions=4, partition_sizes=[1, 1, 1, 1]",
+ //
+ // and checks whether merge_batch for FIRST_VALUE AND LAST_VALUE
+ // works correctly.
+ async fn first_last_multi_partitions(
Review Comment:
This unit test is written to check for whether `merge_batch` method for `first_value` and `last_value` works as expected.
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277717756
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1370,6 +1370,54 @@ mod tests {
)
}
+ /// some mock data to aggregates
+ fn some_data_v2() -> (Arc<Schema>, Vec<RecordBatch>) {
+ // define a schema.
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::UInt32, false),
+ Field::new("b", DataType::Float64, false),
+ ]));
+
+ // define data.
+ (
+ schema.clone(),
+ vec![
+ RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(UInt32Array::from(vec![2, 3, 4, 4])),
+ Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0, 4.0])),
+ ],
+ )
+ .unwrap(),
+ RecordBatch::try_new(
Review Comment:
Since this data is used to test `first_value` and `last_value`. Data is constructed such that first and last value result is at 2nd and 3rd partition. With this construction, we guarantee that expected result received is not accident, but merging works properly (It doesn't depend on the data insertion order).
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278860514
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -270,6 +267,18 @@ pub struct AggregateExec {
required_input_ordering: Option<LexOrderingReq>,
}
+/// Check whether aggregations is in first stage.
+/// It means that, aggregation input is not another aggregation
+/// `merge_batch` method will not be called for these modes.
+fn is_aggregation_first_stage(mode: &AggregateMode) -> bool {
Review Comment:
moved implementation to the method.
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1279865225
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2689,13 +2703,12 @@ SortPreservingMergeExec: [country@0 ASC NULLS LAST]
--SortExec: expr=[country@0 ASC NULLS LAST]
----ProjectionExec: expr=[country@0 as country, FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as fv1, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@2 as fv2]
------AggregateExec: mode=FinalPartitioned, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
---------SortExec: expr=[ts@1 ASC NULLS LAST]
-----------CoalesceBatchesExec: target_batch_size=8192
-------------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
---------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)], ordering_mode=None
-----------------SortExec: expr=[ts@1 ASC NULLS LAST]
-------------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
---------------------MemoryExec: partitions=1, partition_sizes=[1]
+--------CoalesceBatchesExec: target_batch_size=8192
+----------RepartitionExec: partitioning=Hash([country@0], 8), input_partitions=8
+------------AggregateExec: mode=Partial, gby=[country@0 as country], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
Review Comment:
Ah, I forgot that the `first_value` and `last_value` aggregators store the value of the ordering internally https://github.com/apache/arrow-datafusion/blob/504f24767486b8bf9cb08dd54b829b1654f1054f/datafusion/physical-expr/src/aggregate/first_last.rs#L155-L161
Makes sense -- I will review the rest of the changes in this PR carefully tomorrow
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo merged pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo merged PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278556220
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I haven't reviewed this super carefully yet, but `GroupOrderingNone` was substantially faster than` GroupOrderingPartial`, because `GroupOrderingPartial` requires additional comparisons per group key.
Thus I would expect this test to slow down performance. I will run some performance tests next week.
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -270,6 +267,18 @@ pub struct AggregateExec {
required_input_ordering: Option<LexOrderingReq>,
}
+/// Check whether aggregations is in first stage.
+/// It means that, aggregation input is not another aggregation
+/// `merge_batch` method will not be called for these modes.
+fn is_aggregation_first_stage(mode: &AggregateMode) -> bool {
Review Comment:
Maybe this would be easier to find if it were a method on `AggregateMode`
```rust
impl AggregateMode {
...
pub fn is_first_stage(&self) -> bool {....}
}
```
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1278576304
##########
datafusion/core/src/physical_plan/aggregates/order/mod.rs:
##########
@@ -52,7 +52,6 @@ impl GroupOrdering {
} = ordering;
Ok(match mode {
- GroupByOrderMode::None => GroupOrdering::None,
Review Comment:
I should have stressed it, this change, doesn't affect code path for GroupOrderingNone. I recognized that we already store aggregation as option as in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/mod.rs#L269). When this mode is `Option::None`, it is equivalent to `GroupByOrderMode::None`. So I thought, there is no need to keep track of `AggregationMode::None` separately. As can be seen in [link](https://github.com/apache/arrow-datafusion/blob/01d7dbaf392b4a217e1a27c96bbcd1126b763828/datafusion/core/src/physical_plan/aggregates/row_hash.rs#L239). When option is `None`, `GroupOrdering::None ` is used. If this is misleading, I can retract this change. It is not important, for 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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mustafasrepo commented on a diff in pull request #7129: [bug]: Fix multi partition wrong column requirement bug
Posted by "mustafasrepo (via GitHub)" <gi...@apache.org>.
mustafasrepo commented on code in PR #7129:
URL: https://github.com/apache/arrow-datafusion/pull/7129#discussion_r1277720658
##########
datafusion/core/tests/sqllogictests/test_files/groupby.slt:
##########
@@ -2760,8 +2772,8 @@ Projection: FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS
physical_plan
ProjectionExec: expr=[FIRST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@0 as fv1, LAST_VALUE(sales_global.amount) ORDER BY [sales_global.ts ASC NULLS LAST]@1 as fv2]
--AggregateExec: mode=Final, gby=[], aggr=[FIRST_VALUE(sales_global.amount), LAST_VALUE(sales_global.amount)]
-----SortPreservingMergeExec: [ts@0 ASC NULLS LAST]
Review Comment:
Also `SortPreservingMergeExec` is replaced with `CoalescePartitionsExec`. since in final mode we no longer require any ordering for its input.
--
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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org