Fix the lazy fanout test (#31726)

This is fixing lazy fanout failure in new bulk execution backend (#30903).

It turns out we can have a smaller fix (than making Dataset lazy-only: #31668) for lazy fanout.

python/ray/data/_internal/block_list.py

@@ -33,6 +33,9 @@ def __init__(
         # eagerly deleted after read by the consumer.
         self._owned_by_consumer = owned_by_consumer
 
+    def __repr__(self):
+        return f"BlockList(owned_by_consumer={self._owned_by_consumer})"
+
     def get_metadata(self, fetch_if_missing: bool = False) -> List[BlockMetadata]:
         """Get the metadata for all blocks."""
         return self._metadata.copy()

python/ray/data/_internal/lazy_block_list.py

@@ -98,6 +98,9 @@ def __init__(
         self._owned_by_consumer = owned_by_consumer
         self._stats_actor = _get_or_create_stats_actor()
 
+    def __repr__(self):
+        return f"LazyBlockList(owned_by_consumer={self._owned_by_consumer})"
+
     def get_metadata(self, fetch_if_missing: bool = False) -> List[BlockMetadata]:
         """Get the metadata for all blocks."""
         if all(meta is not None for meta in self._cached_metadata):

python/ray/data/tests/test_optimize.py

@@ -257,9 +257,9 @@ def inc(x):
     ray.get(map_counter.reset.remote())
     # The source data shouldn't be cleared since it's non-lazy.
     ds = ray.data.from_items(list(range(10)))
-    # Add extra transformation before being lazy.
-    ds = ds.map(inc)
+    # Add extra transformation after being lazy.
     ds = ds.lazy()
+    ds = ds.map(inc)
     ds1 = ds.map(inc)
     ds2 = ds.map(inc)
     # Test content.