The intermediate type of collect_list/collect_set isn't compatible with Spark

[NEUpanning]

Bug description

The intermediate type of Spark's collect_list/collect_set is BINARY. The intermediate data type for Velox's collect_list/collect_set is ARRAY, which is incompatible with BINARY. The current workaround implemented in Gluten incurs some issues, which include gluten#8227 and gluten#8184.

A complete solution involves changing the intermediate type of Velox's collec_list/collect_set to VARBINARY and using UnsafeArrayData format to do the SerDe.

System information

/

Relevant logs

No response

[NEUpanning]

cc @zhztheplayer

[zhztheplayer]

Thank you for working on this @NEUpanning.

This will involves changes on Spark functions collect_list / collect_set to make them better aligned with vanilla Spark. Could see the links posted in PR description for more information. If anyone has any thoughts about the topic please kindly discuss here.

cc @rui-mo

[NEUpanning]

In the intermediate step of the collect_list function, we need the raw input type to deserialize the intermediate data. However, we cannot obtain this from argTypes or resultType, so we need to include serialized raw input type as part of the intermediate data. After discussing this with @zhztheplayer offline, we concluded that we could use Type::serialize to serialize the type information.

[NEUpanning]

@zhztheplayer @rui-mo
I opened a draft PR in #12121. After this change, Velox will not automatically generate companion
functions for collect_list if the return type is still array(T). Perhaps, we could create a signature for each data type like this:

std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures;
  for (const auto& inputType :
      {"tinyint", "smallint", "integer", "bigint", "real", "double" ./* ... */}) {
    auto returnType = fmt::format("array({})", inputType);
    signatures.push_back(exec::AggregateFunctionSignatureBuilder()
                             .returnType(returnType)
                             .intermediateType("varbinary")
                             .argumentType(inputType)
                             .build());
  }

However, we may not be able to enumerate all types. What do you think? Thanks!