feat(pyspark): pandas agg udf support

[ted0928]

Description of changes

Indroduce new annotation for aggregate udf as @ibis.udf.agg.pandas
Add implement for pyspark

Issues closed

• Resolves feat(pyspark): support udaf  #9173

[cpcloud]

@ted0928 Thanks for the PR!

Unfortunately we're not yet ready to go down the rabbit hole of non-builtin aggregate functions yet.

The way the implementations of custom aggregates work greatly influences their API design (which isn't really true for scalar UDFs).

For example, in all of our backends that support user-defined aggregate functions there is no assumption that all the data for a given aggregate call will be in the same place.

The consequences of not making this assumption are that users typically need to implement at least 3 methods:

1. Some initialization of the state that will be stored and mutated in the aggregate.
2. A "step" method that accepts (conceptually) a row, and updates the state according to the specific aggregate.
3. Some "finalize" method if anything needs to be done to compute the final value of the aggregate.
4. Optionally a method that can back out the changes for the "step" method to support window functions, but even this is pretty dicey since this actually forces a specific algorithm in the database and forces a non-optimal worst-case runtime for window functions.

This is what the SQLite API looks like in both Python and C.

This is also typically what a single-node system looks like.

For a distributed system, typically "step" operates on a partial aggregate, and there's an additional "merge" step that combines the partial aggregate state into another bundle of state.

PySpark Pandas UDFs are an exception (a distributed aggregate that requires none of the complexity of the above, sacrificing reliability and scalability for convenience since the data must all be on a single node for an aggregate like this to work) and so starting on this part of the API sets a precedent for an API whose design is very uncertain at the moment.