[FEA] Support HashAggregate on struct and nested struct

[viadea]

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I wish the RAPIDS Accelerator for Apache Spark would [...]
Support HashAggregate on struct and nested struct.

Currently a group-by on struct can not work on GPU:

  !Exec <ShuffleExchangeExec> cannot run on GPU because Columnar exchange without columnar children is inefficient
    @Partitioning <HashPartitioning> could run on GPU
      @Expression <AttributeReference> name#16 could run on GPU
    !Exec <HashAggregateExec> cannot run on GPU because Nested types in grouping expressions are not supported

Describe the solution you'd like
A clear and concise description of what you want to happen.
GpuHashAggregate can work on nested struct.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
Add any other context, code examples, or references to existing implementations about the feature request here.

[revans2]

I would like to get some clarification on the requirements.

Is this only for the group by keys, or are there requirements for specific aggregations that also need to support nested structs.
What types are allowed in the structs? Specifically are Lists/Arrays allowed to be in the structs? If so then this is going to be much harder, because cudf will likely push back on that.
Do you need a single level for a struct or do you need multiple level deep?

[viadea]

I would like to get some clarification on the requirements.

Is this only for the group by keys, or are there requirements for specific aggregations that also need to support nested structs.
What types are allowed in the structs? Specifically are Lists/Arrays allowed to be in the structs? If so then this is going to be much harder, because cudf will likely push back on that.
Do you need a single level for a struct or do you need multiple level deep?

Here are 2 scenarios to trigger this HashAggregate on struct and/or nested struct.

1. Union on single-level struct + nested struct(2-levels)
   Currently in 21.08 snapshot, GpuUnion does work on struct and also nested struct per my tests.
   However since there is a Union will do de-duplicate, so HashAggregate after the GpuUnion will fallback on CPU.
   So the firstly use case is for Union, and the the minimum reproduce is:

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

val data = Seq(
    Row(Row("Adam ","","Green"),"1","M",1000),
    Row(Row("Bob ","Middle","Green"),"2","M",2000),
    Row(Row("Cathy ","","Green"),"3","F",3000)
)

val schema = (new StructType()
  .add("name",new StructType()
    .add("firstname",StringType)
    .add("middlename",StringType)
    .add("lastname",StringType)) 
  .add("id",StringType)
  .add("gender",StringType)
  .add("salary",IntegerType))

val df = spark.createDataFrame(spark.sparkContext.parallelize(data),schema)
df.write.format("parquet").mode("overwrite").save("/tmp/testparquet")

val df2 = spark.read.parquet("/tmp/testparquet")
val df3 = spark.read.parquet("/tmp/testparquet")

df2.createOrReplaceTempView("df2")
df3.createOrReplaceTempView("df3")

val nested_struct_df=spark.sql("select struct(name, struct(name.firstname, name.lastname) as newname) as col,name from df2")
nested_struct_df.createOrReplaceTempView("nested_struct_df")

//Single-level union falls back on CPU due to HashAggregate
val single_level_struct_union_df=spark.sql("select name from df2 union select name from df3")
single_level_struct_union_df.show
single_level_struct_union_df.explain

//Two-level union falls back on CPU due to HashAggregate
val two_level_struct_union_df=spark.sql("select col from nested_struct_df union select col from nested_struct_df")
two_level_struct_union_df.show
two_level_struct_union_df.explain

2. HashAggregate on the single-level struct with a function as count
   Minimum reproduce is:

spark.sql("select count(*) from df2 group by name").show
spark.sql("select count(*) from df2 group by name").explain

[mythrocks]

I'm working on the libcudf side of this right now. I hope to have a PR for this later this week.

[sameerz]

Related libcudf PR: rapidsai/cudf#9024