fix: support UDAFs with different intermediate schema

[big-andy-coates]

Description

@vpapavas recently enhanced UDAFs to support a different intermediate type, e.g. the AVG UDAF might take an DOUBLE input and output, but it's intermediate type is STRUCT<COUNT BIGINT, SUM DOUBLE>.

This enhancement did not include the necessary parallel changes needed to support the same for static queries.

The rocksdb state stores store the intermediate type. Therefore, static queries, which read data from the rocksdb state stores, must now perform an additional step to convert the intermediate aggregator state to the output type of the aggregator.

How to review

In the future, static queries will also generate physical query plans. But for now they are hacked together in StaticQueryExecutor. So please refrain from commenting on the architecture of this PR. I'm well aware it is less than ideal. Once @rodesai has finished decoupling the KS implementation from the physical plan I can start to address this.

• KsqlMaterialization and it's factory KsqlMaterializationFactory contain the main change, which is to add another step to convert aggregate internal state to final state.
• AggregatesInfo was added to capture the data needed for this new step
• And AggregateNode enhanced to capture it and pass it to MaterializationInfo .

Also fixed some bad import order issues along the way, (or rather my IDE did)

Testing done

materialized-aggregate-static-queries.json has new tests to cover this.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[vpapavas]

LGTM +1.

This is what I understand: We use the KsqlMaterializationFactory to create a new Materialization which is done in the Physical Plan builder and represents materialized state that we store in RocksDB. These materializations are needed to store intermediate state between streams steps. Like for example, when aggregating where intermediate data is stored in RocksDB. You added the map functions of UDAFs to be called so that we can go from the intermediate materialized state creating during aggregation to the final materialized state after aggregation (whose schemas might be different).

[rodesai]

LGTM w/ 1 nit inline