[Datasets] Streaming executor fixes #5

[jianoaix]

Why are these changes needed?

• Fix [Datasets] LazyBlocklist split fails to split heteroeneous list #32950
• Invoke ExecutionPlan.execute() directly (instead of ds.take()) when the purpose is to test ExecutionPlan.execute(): if using ds.take() it may run the new execution backend which doesn't invoke the `ExecutionPlan.execute()'

Related issue number

#32132

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

[clarkzinzow]

Hmm it seems like this cached_metadata should always be homogeneous, i.e. it should always contain a BlockMetadata for each element. Do you know how this heterogeneity is happening?

[jianoaix]

We set only the first block when figuring out the schema, so e.g. with this:

import ray

inputs = ["example://iris.csv"] * 100
ds = ray.data.read_csv(inputs, parallelism=10)
print("before:", ds._plan._in_blocks._cached_metadata)
ds.schema()
print("after:", ds._plan._in_blocks._cached_metadata)
ds._plan._in_blocks.split(2)

It's producing:

before: [None, None, None, None, None, None, None, None, None, None]
after: [[BlockMetadata(num_rows=1500, size_bytes=66500, schema=sepal.length: double
sepal.width: double
petal.length: double
petal.width: double
variety: string, input_files=array(['/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv',
       '/home/ubuntu/ray/python/ray/data/examples/data/iris.csv'],
      dtype='<U55'), exec_stats={'wall_time_s': 0.36291642487049103, 'cpu_time_s': 0.3410033880000001, 'node_id': 'f3e389087180baf4bcde82efe3873d1139be957718e69465786af17d'})], None, None, None, None, None, None, None, None, None]

[ericl]

Could we pull this out into an array split util and use it for the above splits too?

Otherwise it's not clear this is doing the same thing as array split.

[jianoaix]

Sg, done.

[jianoaix]

The purpose of this test is testing ExecutionPlan.execute() so we run it directly. Running ds.take() may invoke the new execution backend when the flag is on.

[jianoaix]

Tests passing (the failures are relevant to this pr).