You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "jayzhan211 (via GitHub)" <gi...@apache.org> on 2023/08/09 09:13:51 UTC
[GitHub] [arrow-datafusion] jayzhan211 opened a new pull request, #7242: Support array aggregate function
jayzhan211 opened a new pull request, #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242
## 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 #7213 .
Ref: #7214 .
## 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?
<!--
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?
<!--
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1672501529
@izveigor @alamb Ready for review, 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.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1924748812
I am trying to go through old PRs and make sure we don't lose any -- this one has not had much activity and has accumulated conflicts. Marking as draft so it isn't on the review queue. Please feel free to reopen / mark as ready for review if it is
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1817791098
> @jayzhan211 I wonder if you could start looking at the LIst / Array code and find ways to make it simpler before we begin implementing new features --
>
> I feel like the code is at a place where doing anything is challenging to get a timely review (because the code is so complicated). Also, when making changes like #7629 results in several regressions, that is a sign to me that the existing code needs to be simplified / improved before we can add significant new features
I think Aggregate is so different that cleanup existing array function does not help so much. Maybe we can move on this again? Let me know if there is any issue we should focus on before moving on ArrayAggregate.
--
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
Re: [PR] Support array aggregate sum function [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #7242: Support array aggregate sum function
URL: https://github.com/apache/datafusion/pull/7242
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1768274394
It is significantly simpler after #7352 :)
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292887025
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1128,18 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator| {
+ let column_value = accumulator.evaluate_v2();
Review Comment:
@alamb
If we want `accumulator.evaluate()` here, we might need to introduce something like `ScalarValue::Array(Option<Vec<ScalarValue>>)`, the `Vec` here is to represent values of rows.
i.e. ScalarValue::Array(vec![1, 2, 3, 4])
-> Int64Array(1,2,3,4)
--
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1676144678
One of the problems I met is I would get NullArray from batch (RecordBatch) if I did not set the target partition to 1.
https://github.com/apache/arrow-datafusion/blob/00627785718d9d98998021bf44585f32c33af3ea/datafusion/core/src/physical_plan/aggregates/no_grouping.rs#L112-L114
CoalescePartitionsExec is the one that sent the batch with NullArray. The default partition seems to be 4. I would need to set `set datafusion.execution.target_partitions = 1;` in `array.slt`. I'm not yet fully understand why it that, not confident on whether this is the correct fix or workaround.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "edmondop (via GitHub)" <gi...@apache.org>.
edmondop commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1807386171
Following up my discussion on #7214 what's the status of this? Are there blockers that need to be removed / other PR that need to be worked on before this one?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1370953905
##########
datafusion/common/src/scalar.rs:
##########
@@ -763,6 +763,42 @@ impl ScalarValue {
}
}
+ // ListArray compatible version of new_primitive
+ pub fn new_primitives<T: ArrowPrimitiveType>(
+ values: Vec<Option<T::Native>>,
+ d: &DataType,
+ ) -> Result<Self> {
+ if values.is_empty() {
+ return d.try_into();
+ }
+
+ // We need to convert it to ScalarValue::Primitive (Int64(0)) instead of ScalarValue::List (List([PrimitiveArray<Int64> [0]]))
Review Comment:
Nice advice. I will keep this in mind
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369431592
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
I'm thinking about introducing `AggregateFunction::ArraySum` for array_aggreate('sum') and creating an accumulator `SumAccumulator` that is specialized for array cases, instead of introducing `ArraySumAccumulator` under existing `AggregateFunction::Sum`
--
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1676386395
I miss the latest comment, let me try out 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.
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 #7242: Support array aggregate function
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1290416046
##########
docs/source/user-guide/sql/scalar_functions.md:
##########
@@ -1495,6 +1496,26 @@ from_unixtime(expression)
- [make_list](#make_list)
- [trim_array](#trim_array)
+### `array_aggregate`
+
+Allows the execution of arbitrary existing aggregate functions on the elements of a list.
Review Comment:
```suggestion
Allows the execution of arbitrary existing aggregate function `name` on the elements of a list.
```
##########
datafusion/core/tests/sqllogictests/test_files/array.slt:
##########
@@ -2361,6 +2361,43 @@ from flatten_table;
[1, 2, 3] [1, 2, 3, 4, 5, 6] [1, 2, 3] [1.0, 2.1, 2.2, 3.2, 3.3, 3.4]
[1, 2, 3, 4, 5, 6] [8] [1, 2, 3] [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]
+# array aggregate function
+## array aggregate
+query I
+select array_aggregate([1, 3, 5, 7], 'sum');
Review Comment:
wow -- this is pretty neat
##########
datafusion/physical-expr/src/array_expressions.rs:
##########
@@ -2150,6 +2150,61 @@ pub fn array_has_all(args: &[ArrayRef]) -> Result<ArrayRef> {
Ok(Arc::new(boolean_builder.finish()))
}
+macro_rules! array_sum_internal(
+ ($ARRAY:expr, $ARRAY_TYPE:ident, $DATA_TYPE:ident) => {{
+ let mut values = vec![];
+ for arr in $ARRAY.iter() {
+ if let Some(arr) = arr {
+ let arr = downcast_arg!(arr, $ARRAY_TYPE);
+ let sum = arr.values().iter().sum::<$DATA_TYPE>();
+ values.push(sum);
+ }
+ }
+ Ok(Arc::new($ARRAY_TYPE::from(values)))
+ }}
+);
+
+/// array_aggregate SQL function
+pub fn array_aggregate(args: &[ArrayRef]) -> Result<ArrayRef> {
+ assert_eq!(args.len(), 2);
+ let func_name = args[1].as_string::<i32>().value(0);
+ let args = &args[0..1];
+ match func_name {
+ "sum" => array_sum(args),
Review Comment:
If you implement this in terms of an [`AggregateExpr`](https://docs.rs/datafusion/latest/datafusion/physical_expr/trait.AggregateExpr.html) and then [Accumulator](https://docs.rs/datafusion/latest/datafusion/logical_expr/trait.Accumulator.html#) you could use the existing aggregate implementations. This would have several benefits:
1. We would avoid code duplication and get all existing aggregates "for free"
2. As new aggregates were added they would also be usable as an array aggregate
3. They would work for non primitive types (e.g. DecimalArray, strings, etc)
The basic idea would be to look up the [aggregate expr](https://docs.rs/datafusion/latest/datafusion/logical_expr/enum.Expr.html#variant.AggregateFunction) during the analysis / parsing phase, create a physical version, and then use it here to instantiate the accumulator
Maybe you could prototype this approach by simply hard coding the mapping `sum` --> [[`AggregateFunction::sum`]](https://docs.rs/datafusion/latest/datafusion/physical_expr/aggregate/build_in/enum.AggregateFunction.html) here and using that to create an `Accumulator` 🤔
##########
datafusion/core/tests/sqllogictests/test_files/array.slt:
##########
@@ -2361,6 +2361,43 @@ from flatten_table;
[1, 2, 3] [1, 2, 3, 4, 5, 6] [1, 2, 3] [1.0, 2.1, 2.2, 3.2, 3.3, 3.4]
[1, 2, 3, 4, 5, 6] [8] [1, 2, 3] [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]
+# array aggregate function
+## array aggregate
+query I
+select array_aggregate([1, 3, 5, 7], 'sum');
+----
+16
+
+## array sum
+query IRI
+select array_sum([1, 3, 5, 7]),
+ array_sum([1.1, 2.2, 3.3]),
+ list_sum([1, -1, 0, 23]);
+----
+16 6.6 23
+
+# TODO: Support nulls in array.
+# query error DataFusion error: This feature is not implemented: Arrays with different types are not supported: \{Int64, Null\}
+# select array_sum([1, null, 3, null]);
Review Comment:
Yes, I agree -- I think if we add a coercion pass that tries to coerce array elements into the same type this will magically start working
##########
datafusion/physical-expr/src/array_expressions.rs:
##########
@@ -2150,6 +2150,61 @@ pub fn array_has_all(args: &[ArrayRef]) -> Result<ArrayRef> {
Ok(Arc::new(boolean_builder.finish()))
}
+macro_rules! array_sum_internal(
+ ($ARRAY:expr, $ARRAY_TYPE:ident, $DATA_TYPE:ident) => {{
+ let mut values = vec![];
+ for arr in $ARRAY.iter() {
+ if let Some(arr) = arr {
+ let arr = downcast_arg!(arr, $ARRAY_TYPE);
+ let sum = arr.values().iter().sum::<$DATA_TYPE>();
+ values.push(sum);
+ }
+ }
+ Ok(Arc::new($ARRAY_TYPE::from(values)))
+ }}
+);
+
+/// array_aggregate SQL function
+pub fn array_aggregate(args: &[ArrayRef]) -> Result<ArrayRef> {
+ assert_eq!(args.len(), 2);
+ let func_name = args[1].as_string::<i32>().value(0);
+ let args = &args[0..1];
+ match func_name {
+ "sum" => array_sum(args),
Review Comment:
If you implement this in terms of an [`AggregateExpr`](https://docs.rs/datafusion/latest/datafusion/physical_expr/trait.AggregateExpr.html) and then [Accumulator](https://docs.rs/datafusion/latest/datafusion/logical_expr/trait.Accumulator.html#) you could use the existing aggregate implementations. This would have several benefits:
1. We would avoid code duplication and get all existing aggregates "for free"
2. As new aggregates were added they would also be usable as an array aggregate
3. They would work for non primitive types (e.g. DecimalArray, strings, etc)
The basic idea would be to look up the [aggregate expr](https://docs.rs/datafusion/latest/datafusion/logical_expr/enum.Expr.html#variant.AggregateFunction) during the analysis / parsing phase, create a physical version, and then use it here to instantiate the accumulator
Maybe you could prototype this approach by simply hard coding the mapping `sum` --> [[`AggregateFunction::sum`]](https://docs.rs/datafusion/latest/datafusion/physical_expr/aggregate/build_in/enum.AggregateFunction.html) here and using that to create an `Accumulator` 🤔
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1819169783
> > @jayzhan211 I wonder if you could start looking at the LIst / Array code and find ways to make it simpler before we begin implementing new features --
> > I feel like the code is at a place where doing anything is challenging to get a timely review (because the code is so complicated). Also, when making changes like #7629 results in several regressions, that is a sign to me that the existing code needs to be simplified / improved before we can add significant new features
>
> @alamb I think ArrayAggregate is so different that cleanup existing array function does not help so much. Maybe we can move on this again? Let me know if there is any issue we should focus on before moving on ArrayAggregate.
My concern is that aggregate functions are some of the most important features in DataFusion as they are widely used and their performance is very important perspective. I am very concerned that any change in how the aggregates operate will cause issues downstream (either functionality or performance) as well as make it harder to maintain.
For this case especially I am worried about adding features for their own sake without anyone who really wants
So one question I have is if anyone is waiting on this feature, and if so perhaps they can help implement / test it.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1399365249
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
@alamb I agree that we should not introduce any downgrade especially for this widely used function.
I think we can either run the comparison that ensure there is no performance downgrade for `sum` or we just consider the initial approach that we build `array_aggreate` like other array function.
In current approach, ArraySum is actually very different from Sum. We have another Accumulator for Array version. I think this is the only place that might effect the performance of current sum aggregate, where we need to differentiate Array and non-Array cases.
Actually, the initial goal that we try to done this in Accumulator "To reduce code duplication" is actually no longer true. Unless we have other reason that we need ArraySum Accumulator. Maybe we should move this back to `array_expression.rs`?
1. Is compare with the performance of Sum enough to ensure there is no downgrade? Not sure what is functionality effect be like.
2. Should we have ArraySum Accumulator for other reason or feature?
3. If we implement array aggregate function in `array_expression.rs` is there any concern for the overall design?
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
@alamb I agree that we should not introduce any downgrade especially for this widely used function.
I think we can either run the comparison that ensure there is no performance downgrade for `sum` or we just consider the initial approach that we build `array_aggreate` like other array function.
In current approach, ArraySum is actually very different from Sum. We have another Accumulator for Array version. I think this is the only place that might effect the performance of current sum aggregate, where we need to differentiate Array and non-Array cases.
Actually, the initial goal that we try to done this in Accumulator "To reduce code duplication" is actually no longer true. Unless we have other reason that we need ArraySum Accumulator. Maybe we should move this back to `array_expression.rs`?
1. Is compare with the performance of Sum enough to ensure there is no downgrade? Not sure what is functionality effect like.
2. Should we have ArraySum Accumulator for other reason or feature?
3. If we implement array aggregate function in `array_expression.rs` is there any concern for the overall design?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1783611442
> @jayzhan211 I wonder if you could start looking at the LIst / Array code and find ways to make it simpler before we begin implementing new features --
>
> I feel like the code is at a place where doing anything is challenging to get a timely review (because the code is so complicated). Also, when making changes like https://github.com/apache/arrow-datafusion/pull/7629 results in several regressions, that is a sign to me that the existing code needs to be simplified / improved before we can add significant new features
Oka
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1807941732
> Following up my discussion on #7214 what's the status of this? Are there blockers that need to be removed / other PR that need to be worked on before this one?
I plan to merge https://github.com/apache/arrow-datafusion/pull/8141 first, then maybe review https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369254508.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1399365249
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
@alamb I agree that we should not introduce any downgrade especially for this widely used function.
I think we can either run the comparison that ensure there is no performance downgrade for `sum` or we just consider the initial approach that we build `array_aggreate` like other array function.
In current approach, ArraySum is actually very different from Sum. We have another Accumulator for Array version. I think this is the only place that might effect the performance of current sum aggregate, where we need to differentiate Array and non-Array cases.
Actually, the initial goal that we try to done this in Accumulator "To reduce code duplication" is actually no longer true. Unless we have other reason that we need ArraySum Accumulator. Maybe we should move this back to `array_expression.rs`?
1. Is comparison the performance of Sum enough to ensure there should be no downgrade?
2. Should we have ArraySum Accumulator for other reason or feature?
3. If we implement array aggregate function in `array_expression.rs` is there any concern for the overall design?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1399365249
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
@alamb I agree that we should not introduce any downgrade especially for this widely used function.
I think we can either run the comparison that ensure there is no performance downgrade for `sum` or we just consider the initial approach that we build `array_aggreate` like other array function.
In current approach, ArraySum is actually very different from Sum. We have another Accumulator for Array version. I think this is the only place that might effect the performance of current sum aggregate, where we need to differentiate Array and non-Array cases.
Actually, the initial goal that we try to done this in Accumulator "To reduce code duplication" is actually no longer true. Unless we have other reason that we need ArraySum Accumulator. Maybe we should move this back to `array_expression.rs`?
1. Is compare with the performance of Sum enough to ensure there is no downgrade?
2. Should we have ArraySum Accumulator for other reason or feature?
3. If we implement array aggregate function in `array_expression.rs` is there any concern for the overall design?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369431592
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
I'm thinking about introducing `AggregateFunction::ArraySum` for array_aggreate('sum') and creating an accumulator for an array based on it.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1370955172
##########
datafusion/common/src/scalar.rs:
##########
@@ -763,6 +763,42 @@ impl ScalarValue {
}
}
+ // ListArray compatible version of new_primitive
+ pub fn new_primitives<T: ArrowPrimitiveType>(
+ values: Vec<Option<T::Native>>,
+ d: &DataType,
+ ) -> Result<Self> {
+ if values.is_empty() {
+ return d.try_into();
+ }
+
+ // We need to convert it to ScalarValue::Primitive (Int64(0)) instead of ScalarValue::List (List([PrimitiveArray<Int64> [0]]))
+ if values.len() == 1 {
+ return Self::new_primitive::<T>(values[0], d);
Review Comment:
Note: I found `List([PrimitiveArray<Int64> [0]])` also represents the same as ScalarValue::Int64(0) in slt file, so I think we can skip this transformation
--
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
Re: [PR] Support array aggregate sum function [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7242:
URL: https://github.com/apache/datafusion/pull/7242#issuecomment-2093950582
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1401931717
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
> If we implement array aggregate function in array_expression.rs is there any concern for the overall design?
Maybe we could put it in array_aggregates.rs (array_expressions.rs is already quite substantial)
Upon further reflection I agree that it makes sense to keep ArraySum separate from the normal Sum aggregator
--
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1674806883
> Thank you @jayzhan211 -- I didn't even know this (array aggregate functions) was a thing!
>
> I think we should consider using the existing aggregate / accumulators rather than reimplementing aggregates for arrays and left some detailed suggestions. Let me know what you think
I also prefer to reuse existing functions, this might not be trivial but I will try to figure it out.
--
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] izveigor commented on a diff in pull request #7242: Support array aggregate function
Posted by "izveigor (via GitHub)" <gi...@apache.org>.
izveigor commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1290550378
##########
datafusion/physical-expr/src/array_expressions.rs:
##########
@@ -2150,6 +2150,61 @@ pub fn array_has_all(args: &[ArrayRef]) -> Result<ArrayRef> {
Ok(Arc::new(boolean_builder.finish()))
}
+macro_rules! array_sum_internal(
+ ($ARRAY:expr, $ARRAY_TYPE:ident, $DATA_TYPE:ident) => {{
+ let mut values = vec![];
+ for arr in $ARRAY.iter() {
+ if let Some(arr) = arr {
+ let arr = downcast_arg!(arr, $ARRAY_TYPE);
+ let sum = arr.values().iter().sum::<$DATA_TYPE>();
+ values.push(sum);
+ }
+ }
+ Ok(Arc::new($ARRAY_TYPE::from(values)))
+ }}
+);
+
+/// array_aggregate SQL function
+pub fn array_aggregate(args: &[ArrayRef]) -> Result<ArrayRef> {
+ assert_eq!(args.len(), 2);
+ let func_name = args[1].as_string::<i32>().value(0);
+ let args = &args[0..1];
+ match func_name {
+ "sum" => array_sum(args),
Review Comment:
Agree with @alamb. Only I want to say that `array_aggregate` function must only accept aggregate functions with **one** argument.
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292887025
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1128,18 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator| {
+ let column_value = accumulator.evaluate_v2();
Review Comment:
@alamb
If we want `accumulator.evaluate()` here, we might need to introduce something like `ScalarValue::Array(Option<Vec<ScalarValue>>)`, the `Vec` here is to represent values of rows.
i.e. ScalarValue::Array(vec![1, 2, 3, 4])
-> Int64Array(1,2,3,4), which is a `four-row` value with (1,2,3,4).
With ScalarValue::List(vec![1,2,3,4])
I would get ListArray(Int64Array(1,2,3,4)), which is the`one-row` value with list[1,2,3,4].
--
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 #7242: Support array aggregate function
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292769515
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1137,37 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator|
+ {
+ println!("mode: {:?}", mode);
+ let res = accumulator.evaluate_v2();
Review Comment:
I wonder if you could do
```suggestion
let res = accumulator.evaluate()?;
res.to_array()
```
--
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 pull request #7242: Support array aggregate function
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1681133109
I am sorry @jayzhan211 -- I don't think I will have time to help work on this PR for the next few days. I will try to find time this wekeend or next week but I am too busy now with TopK and copy related work to help here too. Sorry I am spread too thin 😢
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369434652
##########
datafusion/sql/src/expr/function.rs:
##########
@@ -59,6 +60,26 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
if let Ok(fun) = BuiltinScalarFunction::from_str(&name) {
let args =
self.function_args_to_expr(function.args, schema, planner_context)?;
+
+ // Translate array_aggregate to aggregate function with array argument.
+ if fun == BuiltinScalarFunction::ArrayAggregate {
Review Comment:
If I understand correctly, DataFrame use
https://github.com/apache/arrow-datafusion/blob/eee790f695a58a99e880957d50a33c1f075c8edc/datafusion/expr/src/expr_fn.rs#L153-L161
to call aggregate function. If we need to introduce `AggregateFunction::ArraySum` we just need to use another function that utilizes `ArraySum`, it will have a different path than SQL, right?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1783496898
@jayzhan211 I wonder if you could start looking at the LIst / Array code and find ways to make it simpler before we begin implementing new features --
I feel like the code is at a place where doing anything is challenging to get a timely review (because the code is so complicated). Also, when making changes like https://github.com/apache/arrow-datafusion/pull/7629 results in several regressions, that is a sign to me that the existing code needs to be simplified / improved before we can add significant new features
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1370090137
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
An alternative approach is to go with SumAccumulator or ArraySumAccumulator based on the `data_type` aka ListArray.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1372471567
##########
datafusion/sql/src/expr/function.rs:
##########
@@ -59,6 +60,26 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
if let Ok(fun) = BuiltinScalarFunction::from_str(&name) {
let args =
self.function_args_to_expr(function.args, schema, planner_context)?;
+
+ // Translate array_aggregate to aggregate function with array argument.
+ if fun == BuiltinScalarFunction::ArrayAggregate {
Review Comment:
Let's raise another PR to have a common rewriting rule for SQL, Dataframe, and Expr. I would need more time to deep dive into them.
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1294096414
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -319,11 +392,34 @@ impl Accumulator for SumAccumulator {
fn evaluate(&self) -> Result<ScalarValue> {
// TODO: add the checker for overflow
// For the decimal(precision,_) data type, the absolute of value must be less than 10^precision.
- Ok(self.sum.clone())
+
+ match &self.sum {
+ ColumnarValue::Scalar(sv) => Ok(sv.clone()),
+ ColumnarValue::Array(_) => Err(DataFusionError::Internal(format!(
+ "Sum is not expected to receive the type {:?}",
+ self.sum.data_type()
+ ))),
+ }
+ }
+
+ fn evaluate_v2(&self) -> Result<ColumnarValue> {
+ // TODO: add the checker for overflow
+ // For the decimal(precision,_) data type, the absolute of value must be less than 10^precision.
+
+ match &self.sum {
+ ColumnarValue::Scalar(sv) => Ok(self.sum.clone()),
+ ColumnarValue::Array(ref array) => Ok(self.sum.clone()),
+ }
}
fn size(&self) -> usize {
- std::mem::size_of_val(self) - std::mem::size_of_val(&self.sum) + self.sum.size()
+ match self.sum {
+ ColumnarValue::Scalar(ref sv) => {
+ std::mem::size_of_val(self) - std::mem::size_of_val(sv) + sv.size()
+ }
+ // TODO: Return Correct value
+ ColumnarValue::Array(ref array) => 0,
Review Comment:
Not sure how to get the size of `array: ArrayRef`
--
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 pull request #7242: Support array aggregate function
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1676334979
> Unfortunately, we might need to change the signature in Accumulator from ScalarValue to ColumnarValue. Do you think this signature change is a good decision for accepting the Aggregate function to process Array?
I wonder if you can call `Accumulator::update_batch` instead? It is already possible to go from ColumnarValue --> Array via https://docs.rs/datafusion/latest/datafusion/logical_expr/enum.ColumnarValue.html#method.into_array
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1290311494
##########
datafusion/core/tests/sqllogictests/test_files/array.slt:
##########
@@ -2361,6 +2361,43 @@ from flatten_table;
[1, 2, 3] [1, 2, 3, 4, 5, 6] [1, 2, 3] [1.0, 2.1, 2.2, 3.2, 3.3, 3.4]
[1, 2, 3, 4, 5, 6] [8] [1, 2, 3] [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]
+# array aggregate function
+## array aggregate
+query I
+select array_aggregate([1, 3, 5, 7], 'sum');
+----
+16
+
+## array sum
+query IRI
+select array_sum([1, 3, 5, 7]),
+ array_sum([1.1, 2.2, 3.3]),
+ list_sum([1, -1, 0, 23]);
+----
+16 6.6 23
+
+# TODO: Support nulls in array.
+# query error DataFusion error: This feature is not implemented: Arrays with different types are not supported: \{Int64, Null\}
+# select array_sum([1, null, 3, null]);
Review Comment:
Might be related to https://github.com/apache/arrow-datafusion/issues/7142
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369360626
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -191,12 +196,35 @@ impl<T: ArrowNumericType> Accumulator for SumAccumulator<T> {
Ok(vec![self.evaluate()?])
}
+ // There are two kinds of input, PrimitiveArray and ListArray
+ // ListArray is for multiple-rows input, and PrimitiveArray is for single-row input
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let values = values[0].as_primitive::<T>();
- if let Some(x) = sum(values) {
- let v = self.sum.get_or_insert(T::Native::usize_as(0));
- *v = v.add_wrapping(x);
+ // Wrap single-row input into multiple-rows input and use the same logic as multiple-rows input
+ let list_values = match as_list_array(&values[0]) {
Review Comment:
I don't think I ignored the "error" here. I use the "error" of `as_list_array` to know whether it is List or non-List.
Btw, `Each error requires a string allocation`, can you elaborate more on why there is string allocation?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369251483
##########
datafusion/common/src/scalar.rs:
##########
@@ -763,6 +763,42 @@ impl ScalarValue {
}
}
+ // ListArray compatible version of new_primitive
Review Comment:
```suggestion
/// Return a new `ScalarValue::List` given a `Vec` of primitive values
```
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -191,12 +196,35 @@ impl<T: ArrowNumericType> Accumulator for SumAccumulator<T> {
Ok(vec![self.evaluate()?])
}
+ // There are two kinds of input, PrimitiveArray and ListArray
+ // ListArray is for multiple-rows input, and PrimitiveArray is for single-row input
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let values = values[0].as_primitive::<T>();
- if let Some(x) = sum(values) {
- let v = self.sum.get_or_insert(T::Native::usize_as(0));
- *v = v.add_wrapping(x);
+ // Wrap single-row input into multiple-rows input and use the same logic as multiple-rows input
+ let list_values = match as_list_array(&values[0]) {
Review Comment:
Can we please not ignore the error here? Each error requires a string allocation and this is the performance critical inner loop
##########
datafusion/sql/src/expr/function.rs:
##########
@@ -59,6 +60,26 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
if let Ok(fun) = BuiltinScalarFunction::from_str(&name) {
let args =
self.function_args_to_expr(function.args, schema, planner_context)?;
+
+ // Translate array_aggregate to aggregate function with array argument.
+ if fun == BuiltinScalarFunction::ArrayAggregate {
Review Comment:
If we do the rewrite here, it will only apply to SQL (so `array_aggregate` will not work if it is constructed via the dataframe API or an an `Expr` directly)
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
struct SumAccumulator<T: ArrowNumericType> {
- sum: Option<T::Native>,
+ sum: Vec<Option<T::Native>>,
Review Comment:
I think this change should have a comment that explains what the `Vec` is for and how it is uses (as it now supports ListArrays, each element represents the relative partial sum for the relevant list element, etc
##########
datafusion/common/src/scalar.rs:
##########
@@ -763,6 +763,42 @@ impl ScalarValue {
}
}
+ // ListArray compatible version of new_primitive
+ pub fn new_primitives<T: ArrowPrimitiveType>(
+ values: Vec<Option<T::Native>>,
+ d: &DataType,
+ ) -> Result<Self> {
+ if values.is_empty() {
+ return d.try_into();
+ }
+
+ // We need to convert it to ScalarValue::Primitive (Int64(0)) instead of ScalarValue::List (List([PrimitiveArray<Int64> [0]]))
+ if values.len() == 1 {
+ return Self::new_primitive::<T>(values[0], d);
Review Comment:
I think I would find this behavior very surprising -- that sometimes it returns a ScalarValue::List and sometimes a ScalarValue::Int8, etc -- I realize it is what your changes to sum need, but given it is so specialized, I think it would make more sense to check for a length of 1 at the callsite
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -291,6 +309,27 @@ pub(crate) fn sum_batch(values: &ArrayRef, sum_type: &DataType) -> Result<Scalar
DataType::UInt32 => typed_sum_delta_batch!(values, UInt32Array, UInt32),
DataType::UInt16 => typed_sum_delta_batch!(values, UInt16Array, UInt16),
DataType::UInt8 => typed_sum_delta_batch!(values, UInt8Array, UInt8),
+ DataType::List(field) => {
+ let array = values.as_list::<i32>();
+ let mut scalars: Vec<ScalarValue> = vec![];
+ for arr in array.iter() {
+ if let Some(arr) = arr {
+ let sum = sum_batch(&arr, field.data_type())?;
+ match sum {
+ ColumnarValue::Scalar(sv) => {
+ scalars.push(sv);
+ }
+ e => {
+ return Err(DataFusionError::Internal(format!(
+ "Sum is not expected to receive the type {e:?}"
+ )));
+ }
+ }
+ }
+ }
+ let arr = ScalarValue::iter_to_array(scalars.into_iter())?;
Review Comment:
so in this case, I was thinking `sum` would be called on the values of the `list`, not the list itself 🤔 -- could we call `evaluate` on each scalar 🤔
##########
datafusion/common/src/scalar.rs:
##########
@@ -763,6 +763,42 @@ impl ScalarValue {
}
}
+ // ListArray compatible version of new_primitive
+ pub fn new_primitives<T: ArrowPrimitiveType>(
+ values: Vec<Option<T::Native>>,
+ d: &DataType,
+ ) -> Result<Self> {
+ if values.is_empty() {
+ return d.try_into();
+ }
+
+ // We need to convert it to ScalarValue::Primitive (Int64(0)) instead of ScalarValue::List (List([PrimitiveArray<Int64> [0]]))
Review Comment:
As a suggestion for future PRs, I think comments are best when they describe the *why* of the code (rather than the what) -- in this case I think the why is because this function is doing something very special depending on what mode the accumulator is in.
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
I am very worried about changing how the Sum accumulator works, as the Sum accumulator is one of the performance critical aggregators.
While it would result in duplicated code, I think in this case it would make sense to have a separate accumulator implementation for `array_sum` given this concern
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -291,6 +309,27 @@ pub(crate) fn sum_batch(values: &ArrayRef, sum_type: &DataType) -> Result<Scalar
DataType::UInt32 => typed_sum_delta_batch!(values, UInt32Array, UInt32),
DataType::UInt16 => typed_sum_delta_batch!(values, UInt16Array, UInt16),
DataType::UInt8 => typed_sum_delta_batch!(values, UInt8Array, UInt8),
+ DataType::List(field) => {
+ let array = values.as_list::<i32>();
+ let mut scalars: Vec<ScalarValue> = vec![];
+ for arr in array.iter() {
+ if let Some(arr) = arr {
+ let sum = sum_batch(&arr, field.data_type())?;
+ match sum {
+ ColumnarValue::Scalar(sv) => {
+ scalars.push(sv);
+ }
+ e => {
+ return Err(DataFusionError::Internal(format!(
+ "Sum is not expected to receive the type {e:?}"
+ )));
+ }
+ }
+ }
+ }
+ let arr = ScalarValue::iter_to_array(scalars.into_iter())?;
Review Comment:
so in this case, I was thinking `sum` would be called on the values of the `list`, not the list itself 🤔 -- could we call `evaluate` on each scalar 🤔
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369431592
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
I'm thinking about introducing `AggregateFunction::ArraySum` for array_aggreate('sum') and creating an accumulator based on it, instead of introducing `ArraySumAccumulator` under existing `AggregateFunction::Sum`
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1399365249
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -101,6 +106,43 @@ impl AggregateExpr for Sum {
}
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ if let DataType::List(field) = &self.data_type {
Review Comment:
@alamb I agree that we should not introduce any downgrade especially for this widely used function.
I think we can either run the comparison that ensure there is no performance downgrade for `sum` or we just consider the initial approach that we build `array_aggreate` like other array function.
In current approach, ArraySum is actually very different from Sum. We have another Accumulator for Array version. I think this is the only place that might effect the performance of current sum aggregate, where we need to differentiate Array and non-Array cases.
Actually, the initial goal that we try to done this in Accumulator "To reduce code duplication" is actually no longer true. Unless we have other reason that we need ArraySum Accumulator. Maybe we should move this back to `array_expression.rs`?
1. Is compare with the performance of Sum enough to ensure there should be no downgrade?
2. Should we have ArraySum Accumulator for other reason or feature?
3. If we implement array aggregate function in `array_expression.rs` is there any concern for the overall design?
--
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1676246329
Another big issue, we might need Array for aggregate function Sum so we can process array and return ArrayRef.
I try to process and return ScalarValue, but ScalarValue::List is not a good replacement for ArrayRef. We might need ColumnarValue that can accept either ScalarValue or ArrayRef for `fn sum_batch` and `SumAccumulator`.
https://github.com/apache/arrow-datafusion/blob/00627785718d9d98998021bf44585f32c33af3ea/datafusion/physical-expr/src/aggregate/sum.rs#L269-L300
Unfortunately, we might need to change the signature in `Accumulator` from `ScalarValue` to `ColumnarValue`. Do you think this signature change is a good decision for accepting the Aggregate function to process Array?
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1293374389
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1128,18 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator| {
+ let column_value = accumulator.evaluate_v2();
Review Comment:
It seems not a good idea to introduce `ScalarValue:Array`
https://github.com/apache/arrow-datafusion/blob/2ec0bc106540e4cbebd8b7b7269cbf3fa8ae34e3/datafusion/common/src/scalar.rs#L60-L61
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1769309016
I have this on my review list, and I hope to start working that list down 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] alamb commented on pull request #7242: Support array aggregate function
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1675873211
> I also prefer to reuse existing functions, this might not be trivial but I will try to figure it out.
I can probably find time to help figure out how this might be able to work next week if that would 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.
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] jayzhan211 commented on pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#issuecomment-1676270717
It seems that ColumnarValue for Accumulator works! Just need to confirm this is a reasonable change.
An approach without `set datafusion.execution.target_partitions = 1;` has not been solved.
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292885778
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -291,6 +309,27 @@ pub(crate) fn sum_batch(values: &ArrayRef, sum_type: &DataType) -> Result<Scalar
DataType::UInt32 => typed_sum_delta_batch!(values, UInt32Array, UInt32),
DataType::UInt16 => typed_sum_delta_batch!(values, UInt16Array, UInt16),
DataType::UInt8 => typed_sum_delta_batch!(values, UInt8Array, UInt8),
+ DataType::List(field) => {
+ let array = values.as_list::<i32>();
+ let mut scalars: Vec<ScalarValue> = vec![];
+ for arr in array.iter() {
+ if let Some(arr) = arr {
+ let sum = sum_batch(&arr, field.data_type())?;
+ match sum {
+ ColumnarValue::Scalar(sv) => {
+ scalars.push(sv);
+ }
+ e => {
+ return Err(DataFusionError::Internal(format!(
+ "Sum is not expected to receive the type {e:?}"
+ )));
+ }
+ }
+ }
+ }
+ let arr = ScalarValue::iter_to_array(scalars.into_iter())?;
Review Comment:
This is where I construct Array.
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292885559
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1137,37 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator|
+ {
+ println!("mode: {:?}", mode);
+ let res = accumulator.evaluate_v2();
Review Comment:
The problem is that I need to find a way to represent Array (Column support) with ScalarValue. So that I can get the ArrayRef from ScalarValue `to_array`. ScalarValue::List does not fit at all.
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1292887025
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -1128,10 +1128,18 @@ fn finalize_aggregation(
| AggregateMode::Single
| AggregateMode::SinglePartitioned => {
// merge the state to the final value
- accumulators
+ let res = accumulators
.iter()
- .map(|accumulator| accumulator.evaluate().map(|v| v.to_array()))
- .collect::<Result<Vec<ArrayRef>>>()
+ .map(|accumulator| {
+ let column_value = accumulator.evaluate_v2();
Review Comment:
@alamb
If we want `accumulator.evaluate()` here, we need to introduce something like `ScalarValue::Array(Option<Vec<ScalarValue>>)`, the Vec<> here is to represent values of rows.
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369434652
##########
datafusion/sql/src/expr/function.rs:
##########
@@ -59,6 +60,26 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
if let Ok(fun) = BuiltinScalarFunction::from_str(&name) {
let args =
self.function_args_to_expr(function.args, schema, planner_context)?;
+
+ // Translate array_aggregate to aggregate function with array argument.
+ if fun == BuiltinScalarFunction::ArrayAggregate {
Review Comment:
If I understand correctly, data frame use
https://github.com/apache/arrow-datafusion/blob/eee790f695a58a99e880957d50a33c1f075c8edc/datafusion/expr/src/expr_fn.rs#L153-L161
to call aggregate function. If we need to introduce `AggregateFunction::ArraySum` we just need to use another function that utilizes `ArraySum`, it will have a different path than SQL, right?
--
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
Re: [PR] Support array aggregate sum function [arrow-datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1369431592
##########
datafusion/physical-expr/src/aggregate/sum.rs:
##########
@@ -167,7 +172,7 @@ impl PartialEq<dyn Any> for Sum {
/// This accumulator computes SUM incrementally
Review Comment:
I'm thinking about introducing `AggregateFunction::ArraySum` for array_aggreate('sum') and creating an accumulator `SumAccumulator` that is specialized for array cases, instead of introducing `ArraySumAccumulator` under the existing `AggregateFunction::Sum`.
We also need to do the same to other functions, e.g. `avg`, `min`, `max`
--
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] jayzhan211 commented on a diff in pull request #7242: Support array aggregate function
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #7242:
URL: https://github.com/apache/arrow-datafusion/pull/7242#discussion_r1289432114
##########
datafusion/core/tests/sqllogictests/test_files/array.slt:
##########
@@ -2361,6 +2361,43 @@ from flatten_table;
[1, 2, 3] [1, 2, 3, 4, 5, 6] [1, 2, 3] [1.0, 2.1, 2.2, 3.2, 3.3, 3.4]
[1, 2, 3, 4, 5, 6] [8] [1, 2, 3] [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]
+# array aggregate function
+## array aggregate
+query I
+select array_aggregate([1, 3, 5, 7], 'sum');
+----
+16
+
+## array sum
+query IRI
+select array_sum([1, 3, 5, 7]),
+ array_sum([1.1, 2.2, 3.3]),
+ list_sum([1, -1, 0, 23]);
+----
+16 6.6 23
+
+# TODO: Support nulls in array.
+# query error DataFusion error: This feature is not implemented: Arrays with different types are not supported: \{Int64, Null\}
+# select array_sum([1, null, 3, null]);
Review Comment:
I would like to fix this in the next 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