[core][compiled graphs] streaming support for actors in compiled graphs

[wingkitlee0]

Description

The following actor works with .remote() but not .bind()

@ray.remote
class Mapper:
    def __init__(self, func):
        self.func = func

    def map(self, x: Iterable[Any]) -> Iterator[Any]:
        for item in x:
            yield self.func(item)

map_func = Mapper.remote(lambda x: x * x)

Use case

I was trying to re-implement some Dask functions in Ray, e.g., dask bag.