You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/11/15 01:29:55 UTC
[GitHub] [arrow-datafusion] viirya opened a new pull request #1300: Implement array_agg aggregate function
viirya opened a new pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300
# 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 #1085.
# 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 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] Jimexist commented on pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
Jimexist commented on pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#issuecomment-969079693
> Thanks for the contribution.
>
> I'm not sure if it's better to do in another PR, but both Trino and PostgreSQL supports `order by`.
>
> Do you consider to support it?
>
>
Maybe that can be put into a separate pull request? I agree that it is useful to add
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749719989
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -120,7 +120,17 @@ impl Accumulator for ArrayAggAccumulator {
}
fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.update(states)
+ if states.is_empty() {
+ return Ok(());
+ };
+
+ match &states[0] {
Review comment:
yea, let me add 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
[GitHub] [arrow-datafusion] alamb commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749718863
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -120,7 +120,17 @@ impl Accumulator for ArrayAggAccumulator {
}
fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.update(states)
+ if states.is_empty() {
+ return Ok(());
+ };
+
+ match &states[0] {
Review comment:
I think it would makes sense to assert here that `states.len() == 1`? so we don't (silently) end up ignoring any other items that might be added (accidentally)
--
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 merged pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
alamb merged pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300
--
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] viirya commented on pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#issuecomment-970495646
Thanks @houqp @Jimexist @Dandandan @alamb @capkurmagati !
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749107611
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,244 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given element data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ Ok(Self {
+ array: ScalarValue::List(Some(Box::new(vec![])), Box::new(datatype.clone())),
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.array.clone()])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
Review comment:
yea, I just rely on the default implementation.
--
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] Jimexist commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
Jimexist commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749179964
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,246 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
Review comment:
not consistent with https://github.com/apache/arrow-datafusion/pull/1300/files#diff-726397db785218ffd76148f15fbe33415f8ac1ec5da4d33bcfcf6dd5c54acffdR112
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r748977830
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,252 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::convert::TryFrom;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given element data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let array_type =
+ DataType::List(Box::new(Field::new("item", datatype.clone(), true)));
+ Ok(Self {
+ array: ScalarValue::try_from(&array_type)?,
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.array.clone()])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ let value = &values[0];
+ let mut new_array: Vec<ScalarValue> = vec![];
+
+ let nested_datatype = match self.array {
+ ScalarValue::List(Some(ref accu_array), ref dt) => {
+ new_array.extend(accu_array.clone().into_iter());
Review comment:
I think so. `accu_array` is immutable so I cannot do `push` on `accu_array` here.
--
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] Dandandan commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749032747
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,244 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
Review comment:
I think this could be `Vec<ScalarValue>` instead.
You only need to convert it to a `List` in `evaluate` and `state` leading to easier `update` implementation.
--
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 change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749719648
##########
File path: datafusion/tests/sql.rs
##########
@@ -1281,6 +1281,60 @@ async fn csv_query_approx_count() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn csv_query_array_agg() -> Result<()> {
+ let mut ctx = ExecutionContext::new();
+ register_aggregate_csv(&mut ctx).await?;
+ let sql =
+ "SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 2) test";
Review comment:
can you please add an `ORDER BY` to this query - as written it may be non deterministic (the first two values returned are not guaranteed to be in any order):
```suggestion
"SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 ORDER BY c13 LIMIT 2) test";
```
##########
File path: datafusion/tests/sql.rs
##########
@@ -1281,6 +1281,60 @@ async fn csv_query_approx_count() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn csv_query_array_agg() -> Result<()> {
+ let mut ctx = ExecutionContext::new();
+ register_aggregate_csv(&mut ctx).await?;
+ let sql =
+ "SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 2) test";
+ let actual = execute_to_batches(&mut ctx, sql).await;
+ let expected = vec![
+ "+------------------------------------------------------------------+",
+ "| ARRAYAGG(test.c13) |",
+ "+------------------------------------------------------------------+",
+ "| [6WfVFBVGJSQb7FhA7E0lBwdvjfZnSW, C2GT5KVyOPZpgKVl110TyZO0NcJ434] |",
+ "+------------------------------------------------------------------+",
+ ];
+ assert_batches_eq!(expected, &actual);
+ Ok(())
+}
+
+#[tokio::test]
+async fn csv_query_array_agg_empty() -> Result<()> {
Review comment:
👍
##########
File path: datafusion/tests/sql.rs
##########
@@ -1281,6 +1281,60 @@ async fn csv_query_approx_count() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn csv_query_array_agg() -> Result<()> {
+ let mut ctx = ExecutionContext::new();
+ register_aggregate_csv(&mut ctx).await?;
+ let sql =
+ "SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 2) test";
+ let actual = execute_to_batches(&mut ctx, sql).await;
+ let expected = vec![
+ "+------------------------------------------------------------------+",
+ "| ARRAYAGG(test.c13) |",
+ "+------------------------------------------------------------------+",
+ "| [6WfVFBVGJSQb7FhA7E0lBwdvjfZnSW, C2GT5KVyOPZpgKVl110TyZO0NcJ434] |",
+ "+------------------------------------------------------------------+",
+ ];
+ assert_batches_eq!(expected, &actual);
+ Ok(())
+}
+
+#[tokio::test]
+async fn csv_query_array_agg_empty() -> Result<()> {
+ let mut ctx = ExecutionContext::new();
+ register_aggregate_csv(&mut ctx).await?;
+ let sql =
+ "SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 0) test";
+ let actual = execute_to_batches(&mut ctx, sql).await;
+ let expected = vec![
+ "+--------------------+",
+ "| ARRAYAGG(test.c13) |",
+ "+--------------------+",
+ "| [] |",
+ "+--------------------+",
+ ];
+ assert_batches_eq!(expected, &actual);
+ Ok(())
+}
+
+#[tokio::test]
+async fn csv_query_array_agg_one() -> Result<()> {
+ let mut ctx = ExecutionContext::new();
+ register_aggregate_csv(&mut ctx).await?;
+ let sql =
+ "SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 1) test";
Review comment:
same comment here about needing an `ORDER BY` please
--
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] viirya commented on pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#issuecomment-969141550
Thanks @Jimexist @alamb @capkurmagati Let me add some end to end tests. I will try to add `order by` in a separate PR later.
--
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] Dandandan commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
Dandandan commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749031578
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,244 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given element data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ Ok(Self {
+ array: ScalarValue::List(Some(Box::new(vec![])), Box::new(datatype.clone())),
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.array.clone()])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
Review comment:
Did you leave `update_batch` and `merge_batch` on purpose?
I think at this point it is hard to think of a much more efficient implementation (avoiding converting every item to scalars), given that we don't have columnar storage for aggregates yet.
--
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 change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
alamb commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749718863
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -120,7 +120,17 @@ impl Accumulator for ArrayAggAccumulator {
}
fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
- self.update(states)
+ if states.is_empty() {
+ return Ok(());
+ };
+
+ match &states[0] {
Review comment:
I think it would makes sense to assert here that `states.len() == 1`? so we don't (silently) end up ignoring any other items that might be added (accidentally / erroniously)
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r748982466
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,252 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::convert::TryFrom;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given element data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let array_type =
+ DataType::List(Box::new(Field::new("item", datatype.clone(), true)));
+ Ok(Self {
+ array: ScalarValue::try_from(&array_type)?,
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.array.clone()])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ let value = &values[0];
+ let mut new_array: Vec<ScalarValue> = vec![];
+
+ let nested_datatype = match self.array {
+ ScalarValue::List(Some(ref accu_array), ref dt) => {
+ new_array.extend(accu_array.clone().into_iter());
Review comment:
Ah, I think I can remove the copying. Let me update 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
[GitHub] [arrow-datafusion] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749104394
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,244 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
Review comment:
Good idea. Updated.
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749706449
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,256 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: Vec<ScalarValue>,
+ datatype: DataType,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given item data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ Ok(Self {
+ array: vec![],
+ datatype: datatype.clone(),
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![ScalarValue::List(
+ Some(Box::new(self.array.clone())),
+ Box::new(self.datatype.clone()),
+ )])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ let value = &values[0];
+ self.array.push(value.clone());
+
+ Ok(())
+ }
+
+ fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {
Review comment:
BTW, after we changed the state from a `ScalarValue` to a `Vec<ScalarValue>` in early commit, we also need to update how `merge` works. It cannot call `update` directly now. Found this when adding e2e tests.
--
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] viirya commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749536991
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,246 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
Review comment:
thanks for catching this. should be `item` to be consistent.
--
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] houqp commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
houqp commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r748976403
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,252 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::convert::TryFrom;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ )])
+ }
+
+ fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+ vec![self.expr.clone()]
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+ array: ScalarValue,
+}
+
+impl ArrayAggAccumulator {
+ /// new array_agg accumulator based on given element data type
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let array_type =
+ DataType::List(Box::new(Field::new("item", datatype.clone(), true)));
+ Ok(Self {
+ array: ScalarValue::try_from(&array_type)?,
+ })
+ }
+}
+
+impl Accumulator for ArrayAggAccumulator {
+ fn state(&self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.array.clone()])
+ }
+
+ fn update(&mut self, values: &[ScalarValue]) -> Result<()> {
+ let value = &values[0];
+ let mut new_array: Vec<ScalarValue> = vec![];
+
+ let nested_datatype = match self.array {
+ ScalarValue::List(Some(ref accu_array), ref dt) => {
+ new_array.extend(accu_array.clone().into_iter());
Review comment:
does this require a full memory copy from `accu_array` into `new_array` on every update call?
--
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] Jimexist commented on a change in pull request #1300: Implement array_agg aggregate function
Posted by GitBox <gi...@apache.org>.
Jimexist commented on a change in pull request #1300:
URL: https://github.com/apache/arrow-datafusion/pull/1300#discussion_r749180174
##########
File path: datafusion/src/physical_plan/expressions/array_agg.rs
##########
@@ -0,0 +1,246 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Defines physical expressions that can evaluated at runtime during query execution
+
+use super::format_state_name;
+use crate::error::Result;
+use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
+use crate::scalar::ScalarValue;
+use arrow::datatypes::{DataType, Field};
+use std::any::Any;
+use std::sync::Arc;
+
+/// ARRAY_AGG aggregate expression
+#[derive(Debug)]
+pub struct ArrayAgg {
+ name: String,
+ input_data_type: DataType,
+ expr: Arc<dyn PhysicalExpr>,
+}
+
+impl ArrayAgg {
+ /// Create a new ArrayAgg aggregate function
+ pub fn new(
+ expr: Arc<dyn PhysicalExpr>,
+ name: impl Into<String>,
+ data_type: DataType,
+ ) -> Self {
+ Self {
+ name: name.into(),
+ expr,
+ input_data_type: data_type,
+ }
+ }
+}
+
+impl AggregateExpr for ArrayAgg {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn field(&self) -> Result<Field> {
+ Ok(Field::new(
+ &self.name,
+ DataType::List(Box::new(Field::new(
+ "element",
+ self.input_data_type.clone(),
+ true,
+ ))),
+ false,
+ ))
+ }
+
+ fn create_accumulator(&self) -> Result<Box<dyn Accumulator>> {
+ Ok(Box::new(ArrayAggAccumulator::try_new(
+ &self.input_data_type,
+ )?))
+ }
+
+ fn state_fields(&self) -> Result<Vec<Field>> {
+ Ok(vec![Field::new(
+ &format_state_name(&self.name, "array_agg"),
+ DataType::List(Box::new(Field::new(
+ "item",
Review comment:
maybe have a different name?
--
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