[streaming] Aggregation rework to support field expression based aggregations for Pojo data streams