What happened + What you expected to happen

As a followup to #34705, implement pushdown for Limit operators, i.e. move the Limit operator directly after its first upstream Read or AllToAll operator.

Versions / Dependencies

Ray master

Reproduction script

For example,

import ray
ds = ray.data.range(100, parallelism=100).map(lambda x: x).limit(1).materialize()
assert str(ds._plan._logical_plan.dag) == "Read[Read] -> Limit[Limit] -> MapRows[MapRows]", ds._plan._logical_plan.dag

should shutdown after reading the first block.

Issue Severity

None