Adding support for aggregating or buffering streams based on predicate