GH-32884: [C++] Add ordered aggregation

[rtpsw]

This PR implements "Segmented Aggregation" to the existing aggregation node to improve aggregation on ordered data.

A segment group is defined as "a continuous chunk of data that have the same segment key value. e.g, if the input data looks like

[0, 0, 0, 1, 2, 2] 

Then there are three segments [0, 0, 0][1][2, 2]

(Note the "group" in "segment group" here is added to differentiate from "segment", which is defined as "a continuous chunk of data with in a ExecBatch")

Segment aggregation can be used to replace existing hash aggregation in the case that data are ordered. The benefit of this is
(1) We can output aggregation result earlier (as soon as a segment group is fully consumed).
(2) We only need to hold partial aggregation for one segment group to reduce memory usage.

See https://issues.apache.org/jira/browse/ARROW-17642

Replaces #14352

• Closes: [C++] Add ordered aggregation #32884

Follow ups

• [C++] Optimize ordered aggregation #34475
• [C++] Clean up invariant checks in RowSegmenter to be DCHECK #34529

[github-actions]

• Closes: [C++] Add ordered aggregation #32884

[github-actions]

⚠️ GitHub issue #32884has been automatically assigned in GitHub to PR creator.

[rtpsw]

@icexelloss, please review the design in this commit.

@westonpace, in conflict resolution, I see cpp/src/arrow/compute/exec/aggregate.{h,cc} have been deleted. What was the story around that? I'm trying to figure out how best to resolve.

[rtpsw]

@westonpace, in conflict resolution, I see cpp/src/arrow/compute/exec/aggregate.{h,cc} have been deleted. What was the story around that? I'm trying to figure out how best to resolve.

I found the commit you made which removed these files and I generally figured it out. I pushed a resolution of the conflicts.

[rtpsw]

See segmented aggregation as a generalization of ordered aggregation (in the PR replaced by this one).

[icexelloss]

See segmented aggregation as a generalization of ordered aggregation (in the PR replaced by this one).

@rtpsw Can you add some code comment explain the concept, data structure and algorithm? This way the code is more documentation and the reader doesn't need to jump through review/pr links to understand the code.

[westonpace]

Let's remove the mutex stuff for now since we expect this to be single threaded. We can put it back in later.

Let's remove the paths handling chunked arrays as they should no longer be needed.

I think we can even make this simpler by removing all use of ExecSpan in favor of ExecBatch but we can do that in a follow-up (especially since you didn't introduce this).

I'll take another pass once that's done but I went through it pretty thoroughly today. I think I finally understand everything and it looks like a pretty good approach.

Once we move to a multithreaded approach I think the only thing we will need to serialize on is figuring out the segment boundaries. The way this is setup today that responsibility lies with the segmenter so I think this should be pretty straightforward. We can worry / talk more about that later.

[westonpace]

Also, can we add at least one or two basic end-to-end tests in plan_test.cc (or you could create an aggregate_node_test.cc). Partly as examples for future readers as much as anything.

[rtpsw]

Also, can we add at least one or two basic end-to-end tests in plan_test.cc (or you could create an aggregate_node_test.cc). Partly as examples for future readers as much as anything.

My recent commit should have the requested fixes except the above one for adding tests (TBD).

[westonpace]

Minor nit: This example could be slightly improved I think if you used [A, A, B, A] so that readers could see that the segmented group by still does segment.

[westonpace]

Suggested change

	//Handle the input batch
	//If a segment is closed by this batch, then we output the aggregation for the segment
	//If a segment is not closed by this batch, then we add the batch to the segment
	//Extract segments from a batch and run the given handler on them. Note that the
	//handle may be called on open segments which are not yet finished. Typically a
	//handler should accumulate those open segments until a closed segment is reached.

[westonpace]

Suggested change

	Status HandleSegments(std::unique_ptr<RowSegmenter>& segmenter, const ExecBatch& batch,
	Status HandleSegments(RowSegmenter* segmenter, const ExecBatch& batch,

Prefer pointer over mutable reference.

[icexelloss]

Looks like some of the comments in here are not addressed. I will create a follow up to track.

[icexelloss]

nvm - looks like github diff issue

[westonpace]

Suggested change

	///
	/// See also doc in `aggregate_node.cc`

This documentation is for users. I'm not sure we should be directing users to aggregate_node.cc. Also, it's not clear what doc this is referring to. I think this is fine as it is without the "see also".

[icexelloss]

@rtpsw Looks like you missed out on this comment (minor issue)

[rtpsw]

@westonpace, is this good to go?

[westonpace]

From my perspective, yes. We can merge once @icexelloss approves.

[icexelloss]

@rtpsw there are many follow up items from this PR, can you include the list of follows up in PR title so we have at least some ways to track it? If you have a GH issue, please list follow ups here as well.

[icexelloss]

@rtpsw I think this is getting close but there are still a number of unresolved thread. Please check those are resolved and ping me when it is ready for me to take another look.

[rtpsw]

@rtpsw I think this is getting close but there are still a number of unresolved thread. Please check those are resolved and ping me when it is ready for me to take another look.

@icexelloss, I went over the unresolved discussions and commented; I think they can now be resolved. The deferred issues I found are in #34475.

[icexelloss]

Thanks @rtpsw I will take a look later Today.

Can you gather all the follow up issues and put them as a list in the PR description and the origin GH issue as well?

[github-actions]

⚠️ GitHub issue #32884has been automatically assigned in GitHub to PR creator.

[rtpsw]

Can you gather all the follow up issues and put them as a list in the PR description and the origin GH issue as well?

Linked in both PR and original GH issue to #34475 which has the list.

[icexelloss]

Linked in both PR and original GH issue to #34475 which has the list.

Can you put the follow up GH issue link in the list?

[github-actions]

⚠️ GitHub issue #32884has been automatically assigned in GitHub to PR creator.

[icexelloss]

LGTM approved

[icexelloss]

I also edited the PR description to make the purpose of this PR a bit more clear.

[ursabot]

Benchmark runs are scheduled for baseline = 4c05a3b and contender = 9baefea. 9baefea is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished ⬇️0.0% ⬆️0.0%] ec2-t3-xlarge-us-east-2
[Finished ⬇️0.39% ⬆️0.03%] test-mac-arm
[Finished ⬇️0.77% ⬆️0.0%] ursa-i9-9960x
[Finished ⬇️0.69% ⬆️0.0%] ursa-thinkcentre-m75q
Buildkite builds:
[Finished] 9baefea1 ec2-t3-xlarge-us-east-2
[Finished] 9baefea1 test-mac-arm
[Finished] 9baefea1 ursa-i9-9960x
[Finished] 9baefea1 ursa-thinkcentre-m75q
[Finished] 4c05a3b4 ec2-t3-xlarge-us-east-2
[Finished] 4c05a3b4 test-mac-arm
[Finished] 4c05a3b4 ursa-i9-9960x
[Finished] 4c05a3b4 ursa-thinkcentre-m75q
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java