Partition stream by condition

[ngjaying]

Provide a way to partition a stream by condition so that we can calculate independently. For example, if a stream contains data from all devices, we can partition by device id to calculate states for each device in one rule.