[FLINK-1110] Add iteration aggregators to collection based execution