Implement `array_agg` aggregate function #1300

viirya · 2021-11-15T01:29:40Z

Which issue does this PR close?

Closes #1085.

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

datafusion/src/physical_plan/expressions/array_agg.rs

Dandandan · 2021-11-15T06:19:07Z

datafusion/src/physical_plan/expressions/array_agg.rs

+        Ok(vec![self.array.clone()])
+    }
+
+    fn update(&mut self, values: &[ScalarValue]) -> Result<()> {


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.

yea, I just rely on the default implementation.

Dandandan · 2021-11-15T06:22:14Z

datafusion/src/physical_plan/expressions/array_agg.rs

+
+#[derive(Debug)]
+pub(crate) struct ArrayAggAccumulator {
+    array: ScalarValue,


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.

Good idea. Updated.

Dandandan

Looks great! @viirya
I have one minor comment about the accumulator state.

jimexist · 2021-11-15T10:15:27Z

datafusion/src/physical_plan/expressions/array_agg.rs

+        Ok(Field::new(
+            &self.name,
+            DataType::List(Box::new(Field::new(
+                "element",


not consistent with https://github.com/apache/arrow-datafusion/pull/1300/files#diff-726397db785218ffd76148f15fbe33415f8ac1ec5da4d33bcfcf6dd5c54acffdR112

thanks for catching this. should be item to be consistent.

jimexist · 2021-11-15T10:15:44Z

datafusion/src/physical_plan/expressions/array_agg.rs

+        Ok(vec![Field::new(
+            &format_state_name(&self.name, "array_agg"),
+            DataType::List(Box::new(Field::new(
+                "item",


maybe have a different name?

jimexist

thanks for the pull request. can you add sql test and psql comparison integration tests?

i'd like to cover 0-/1-/2-element cases and also with a group by to see how it's handled differently

alamb

Thank you @liukun4515 -- I agree with @jimexist that having the end to end tests (aka showing that one can invoke array_agg from SQL) is quite important. Otherwise this PR is 👍 to me

capkurmagati

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?

jimexist · 2021-11-15T16:24:03Z

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

viirya · 2021-11-15T17:28:34Z

Thanks @jimexist @alamb @capkurmagati Let me add some end to end tests. I will try to add order by in a separate PR later.

viirya · 2021-11-15T21:48:15Z

datafusion/src/physical_plan/expressions/array_agg.rs

+        Ok(())
+    }
+
+    fn merge(&mut self, states: &[ScalarValue]) -> Result<()> {


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.

alamb · 2021-11-15T22:09:35Z

datafusion/src/physical_plan/expressions/array_agg.rs

+            return Ok(());
+        };
+
+        match &states[0] {


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)

yea, let me add it.

alamb · 2021-11-15T22:10:55Z

datafusion/tests/sql.rs

+    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";


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):

Suggested change

"SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 2) test";

"SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 ORDER BY c13 LIMIT 2) test";

alamb · 2021-11-15T22:11:01Z

datafusion/tests/sql.rs

+}
+
+#[tokio::test]
+async fn csv_query_array_agg_empty() -> Result<()> {


alamb · 2021-11-15T22:11:16Z

datafusion/tests/sql.rs

+    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";


same comment here about needing an ORDER BY please

alamb

Looks good -- thank you @viirya !

* Implement array_agg aggregate function. * Avoid copying. * Fix clippy. * For review comment. * Add e2e tests. * Add assert and order by.

viirya · 2021-11-16T17:24:42Z

Thanks @houqp @jimexist @Dandandan @alamb @capkurmagati !

github-actions bot added the datafusion Changes in the datafusion crate label Nov 15, 2021

viirya force-pushed the issue_1085 branch from 64d49ad to c84f75d Compare November 15, 2021 01:43

github-actions bot added the ballista label Nov 15, 2021

houqp added the enhancement New feature or request label Nov 15, 2021

houqp requested review from jimexist, alamb, jorgecarleitao and Dandandan November 15, 2021 02:01

viirya force-pushed the issue_1085 branch from c84f75d to 9d01676 Compare November 15, 2021 02:03

Implement array_agg aggregate function.

23f48a7

viirya force-pushed the issue_1085 branch from 9d01676 to 23f48a7 Compare November 15, 2021 02:13

houqp reviewed Nov 15, 2021

View reviewed changes

datafusion/src/physical_plan/expressions/array_agg.rs Outdated Show resolved Hide resolved

Avoid copying.

9ca4543

viirya force-pushed the issue_1085 branch from dda10c1 to 9ca4543 Compare November 15, 2021 03:30

Fix clippy.

a75a84c

houqp approved these changes Nov 15, 2021

View reviewed changes

Dandandan reviewed Nov 15, 2021

View reviewed changes

Dandandan requested changes Nov 15, 2021

View reviewed changes

For review comment.

ab908a8

jimexist reviewed Nov 15, 2021

View reviewed changes

Dandandan approved these changes Nov 15, 2021

View reviewed changes

alamb reviewed Nov 15, 2021

View reviewed changes

capkurmagati reviewed Nov 15, 2021

View reviewed changes

Add e2e tests.

bc107c5

viirya commented Nov 15, 2021

View reviewed changes

alamb reviewed Nov 15, 2021

View reviewed changes

Add assert and order by.

2d5b367

jimexist approved these changes Nov 16, 2021

View reviewed changes

houqp requested a review from alamb November 16, 2021 06:08

alamb approved these changes Nov 16, 2021

View reviewed changes

alamb merged commit 77c68b2 into apache:master Nov 16, 2021

james727 mentioned this pull request Dec 31, 2021

Implement set_agg aggregate function #1511

Closed

alamb changed the title ~~Implement array_agg aggregate function~~ Implement array_agg aggregate function Feb 10, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement `array_agg` aggregate function #1300

Implement `array_agg` aggregate function #1300

viirya commented Nov 15, 2021

Dandandan Nov 15, 2021

viirya Nov 15, 2021

Dandandan Nov 15, 2021

viirya Nov 15, 2021

Dandandan left a comment

jimexist Nov 15, 2021

viirya Nov 15, 2021

jimexist Nov 15, 2021

jimexist left a comment

alamb left a comment

capkurmagati left a comment

jimexist commented Nov 15, 2021

viirya commented Nov 15, 2021

viirya Nov 15, 2021

alamb Nov 15, 2021 •

edited

Loading

viirya Nov 15, 2021

alamb Nov 15, 2021

alamb Nov 15, 2021

alamb Nov 15, 2021

alamb left a comment

viirya commented Nov 16, 2021

	"SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 LIMIT 2) test";
	"SELECT array_agg(c13) FROM (SELECT * FROM aggregate_test_100 ORDER BY c13 LIMIT 2) test";

Implement array_agg aggregate function #1300

Implement array_agg aggregate function #1300

Conversation

viirya commented Nov 15, 2021

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Dandandan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jimexist left a comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

capkurmagati left a comment

Choose a reason for hiding this comment

jimexist commented Nov 15, 2021

viirya commented Nov 15, 2021

Choose a reason for hiding this comment

alamb Nov 15, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

viirya commented Nov 16, 2021

Implement `array_agg` aggregate function #1300

Implement `array_agg` aggregate function #1300

alamb Nov 15, 2021 •

edited

Loading