[Datasets] Use read stage name for naming Data-read tasks on Ray Dashboard

[scottjlee]

Why are these changes needed?

Previously, the name of read tasks uses the generic DoRead name:

This PR updates the naming so that we use the underlying read stage name, if available from the input LazyBlockList, as the resulting MapOperator; otherwise, we fall back to the existing DoRead name. For example, when running this code:

import ray

ctx = ray.data.DataContext.get_current()
ctx.new_execution_backend = True

ds = ray.data.read_parquet("example://iris.parquet")
data = ds.take(5)
import time
time.sleep(20)

This generates the task with the name ReadParquet:

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[ericl]

Is there a way we can test this?

[c21]

yeah let's also add a unit test if it's not too hard.

[pcmoritz]

Very nice!

[c21]

Documentation failure is not related to this PR. Merging to master.