[FLINK-1110] Add Collection-Based execution for Reduce Operators

Also fix some bugs resulting from moving stuff between packages