[FEA] collect_set on struct[Array]

[viadea]

I wish collect_set can support struct[Array] as input column.
The ask is mainly for struct[Array(String)] and struct[Array(Long)].

For example:

import org.apache.spark.sql.types._
val arrayData = Seq(
    Row("John",List("apple","orange","banana"),1,List(100L,200L,300L)),
    Row("David",List("apple","orange","banana"),2,List(100L,200L,300L)),
    Row("Harry",List("apple","other"),1,List(100L,200L,300L))
)

val arraySchema = new StructType().add("name",StringType).add("fruits", ArrayType(StringType)).add("favorite",IntegerType).add("arraylong", ArrayType(LongType))

val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayData),arraySchema)
df.write.mode("overwrite").format("parquet").save("/tmp/testparquet")
val df2=spark.read.parquet("/tmp/testparquet")
df2.createOrReplaceTempView("df2")

val sqltext = """
select collect_set( CASE WHEN favorite IN (0,1,2,3) THEN struct(name,cast(favorite as string),fruits, arraylong) END) from df2 group by name
"""

spark.sql(sqltext).show()

The not-supported-messages:

          !Expression <CollectSet> collect_set(CASE WHEN favorite#115 IN (0,1,2,3) THEN struct(name, name#113, col2, cast(favorite#115 as string), fruits, fruits#114, arraylong, arraylong#116) END, 0, 0) cannot run on GPU because input expression CaseWhen CASE WHEN favorite#115 IN (0,1,2,3) THEN struct(name, name#113, col2, cast(favorite#115 as string), fruits, fruits#114, arraylong, arraylong#116) END (child ArrayType(StringType,true) is not supported, child ArrayType(LongType,true) is not supported); expression CollectSet collect_set(CASE WHEN favorite#115 IN (0,1,2,3) THEN struct(name, name#113, col2, cast(favorite#115 as string), fruits, fruits#114, arraylong, arraylong#116) END, 0, 0) produces an unsupported type ArrayType(StructType(StructField(name,StringType,true),StructField(col2,StringType,true),StructField(fruits,ArrayType(StringType,true),true),StructField(arraylong,ArrayType(LongType,true),true)),false)

[sameerz]

Depends on PR rapidsai/cudf#10730 from issue rapidsai/cudf#10508 and rapidsai/cudf#10883.

[sameerz]

Depends on rapidsai/cudf#10870

[ttnghia]

A note for cudf implementation: The current cudf unit tests for drop_list_duplicates use hard-coded sorted lists as the expected lists for comparison with the output. When switched to use the new row comparator for nested types, the list elements will not be sorted thus we need to update unit tests as well.

[ttnghia]

Now depends on:

• Refactor collect_set to use cudf::distinct and cudf::lists::distinct rapidsai/cudf#11228

which in turn depends on:

• Implement lists::distinct and cudf::detail::stable_distinct rapidsai/cudf#11149

[NVnavkumar]

Fixed via #6079