ARROW-9679: [Rust] [DataFusion] More efficient creation of final batch from HashAggregateExec

[andygrove]

Instead of walking through the map containing accumulators multiple times (once per grouping expression and once per aggregate expression) let's just walk through it once! Safer and faster.

Other changes in this PR:

• min and max had the wrong data type. MIN(expr) and MAX(expr) should always have the same data type as the underlying expression. They cannot overflow.
• I shortened the name of one of the existing macros so that match statements would fit on a single line.
• Improved error handling in MergeExec so that the reason for threads failing is now propagated in an Err
• Removed unused MapStruct struct

[andygrove]

@alamb @jorgecarleitao @houqp fyi, since you've all been contributing to DataFusion lately

[github-actions]

https://issues.apache.org/jira/browse/ARROW-9679

[alamb]

I reviewed the logic carefully and I didn't see any issues. My only comment that is in the "really should fix before merging" category is removing the println!. The rest are stylistic.

Otherwise I think this PR is looking really good. Nice work @andygrove

[alamb]

I agree that the type of Max and Min should be the same as the input (as those functions are never going to overflow the value type of their input, the way Sum or Avg could.

I actually think the change results in cleaner code too, which is a nice bonus

[alamb]

I think the names $BUILDER_TY and $SCALAR_TY are easier to understand than the previous $TY:ident, $TY2:ty 👍

[alamb]

Suggested change

println!("downcast {:#?} to {:#?}", $BUILDER, $VALUE);

[alamb]

Suggested change

	Some(_) => panic!(),
	Some(_) => Err(ExecutionError::ExecutionError("unexpected type when creating aggregate value".to_string()),

[andygrove]

Thanks. Fixed.

[alamb]

Suggested change

	for i in 0..num_group_expr + num_aggr_expr {
	for i in 0..output_schema.fields().len() {

This is just a style suggestion / defensive coding suggestion (as output_schema.field(i) is matched below).

If you wanted to get all rusty / functional, you could also think about rewriting this as a map over fields. Something like this (untested):

let builders = output_schema
  .fields()
  .iter()
  .map(|f| { match f.data_type ... // the match statement below})
  .collect::<Result<Vec<_>>?;

[andygrove]

I took a quick look at doing this but I needed an explicit assignment still otherwise it complained about the match arms returning different types, so it ended up not being much cleaner really. However, I did change it to use for data_type in &output_types which is a little cleaner.

[alamb]

I am sure you have a good reason, but I didn't see quite why this match arm can't use append_aggr_value! as well

[andygrove]

The issue here was that ScalarValue::Utf8 contains String and the builder wants &str. In all other cases the scalar and builder types are the same.

[alamb]

that might be worth adding in a comment so future readers would not have to wonder.

[andygrove]

Done.

[alamb]

This is another good change (don't panic if there is a thread error while merging) -- maybe worth a mention in the PR title (or maybe even its own PR).

[andygrove]

Thanks for the review @alamb. I've addressed the formatting, removed debug printlns, and tests are now passing.

[andygrove]

@nevi-me @paddyhoran PTAL if you have time.

[jorgecarleitao]

LGTM! Thanks a lot for this, nice cleanup! I closed #7687 in favor of this one as the overall is too high to salvage.