[DBG]

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>

python/ray/data/_internal/output_buffer.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+import pyarrow
+
 from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
 from ray.data.block import Block, BlockAccessor, DataBatch
 from ray.data.context import MAX_SAFE_BLOCK_SIZE_FACTOR
@@ -101,6 +103,9 @@ def next(self) -> Block:
                     target_num_rows, block.num_rows(), copy=False
                 )
 
+        if isinstance(block_to_yield, pyarrow.Table):
+            print(f">>> [DBG] yielding block: {block_to_yield.num_rows}, remaining block: {block_remainder.num_rows if block_remainder else -1}")
+
         self._buffer = DelegatingBlockBuilder()
         if block_remainder is not None:
             self._buffer.add_block(block_remainder)

python/ray/data/tests/test_dynamic_block_split.py

@@ -501,6 +501,9 @@ def test_block_slicing(
         ),
         override_num_blocks=num_tasks,
     ).materialize()
+
+    print(f">>> [DBG] test_block_slicing {ds._plan}")
+
     assert ds._plan.initial_num_blocks() == expected_num_blocks
 
     block_sizes = []