[data] fix progress bar total not updated for streaming_split

[raulchen]

Why are these changes needed?

Fix the issue that streaming_split's total number of outputs are not updated in progress bar.
streaming_split doesn't have _estimated_output_blocks as the map does. but its num_outputs_total will inherit from the previous map.

In the following example, progress bar total should be updated to 10k.

import ray
import time
import numpy as np


ctx = ray.data.DataContext.get_current()
ctx.target_max_block_size = 10 * 1024 * 1024

ds = ray.data.range(1000, parallelism=1000)

def map_fn(_):
    for i in range(10):
        time.sleep(0.1)
        yield {
            "i": [np.ones(10 * 1024 * 1024, dtype=np.uint8)]
        }

ds = ds.map_batches(map_fn, batch_size=None)

split = ds.streaming_split(1)[0]

for _ in split.iter_batches(batch_size=None):
    pass

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 :(

[omatthew98]

Nice quick fix! Verified locally lgtm with your script.

[c21]

LG