class_docstr: Generates output blocks of a given size given a stream of inputs.
  
  This class is used to turn a stream of items / blocks of arbitrary size
  into a stream of blocks of ``target_max_block_size``. The caller should
  check ``has_next()`` after each ``add()`` call, and call ``next()`` to get
  the next block when ``has_next()`` returns True.
  
  When all items have been added, the caller must call ``finalize()`` and
  then check ``has_next()`` one last time.
  
  Examples:
      >>> from ray.data._internal.output_buffer import BlockOutputBuffer
      >>> udf = ... # doctest: +SKIP
      >>> generator = ... # doctest: +SKIP
      >>> # Yield a stream of output blocks.
      >>> output = BlockOutputBuffer(udf, 500 * 1024 * 1024) # doctest: +SKIP
      >>> for item in generator(): # doctest: +SKIP
      ...     output.add(item) # doctest: +SKIP
      ...     if output.has_next(): # doctest: +SKIP
      ...         yield output.next() # doctest: +SKIP
      >>> output.finalize() # doctest: +SKIP
      >>> if output.has_next() # doctest: +SKIP
      ...     yield output.next() # doctest: +SKIP
docstr: Returns the next complete output block.
code: def next(self) -> Block:
      """Returns the next complete output block."""
      assert self.has_next()
  
      block_to_yield = self._buffer.build()
      block_remainder = None
      block = BlockAccessor.for_block(block_to_yield)
      if (
          block.size_bytes()
          >= MAX_SAFE_BLOCK_SIZE_FACTOR * self._target_max_block_size
      ):
          # Slice a block to respect the target max block size.  We only do
          # this if we are more than 50% above the target block size, because
          # this ensures that the last block produced will be at least half
          # the block size.
          num_bytes_per_row = block.size_bytes() // block.num_rows()
          target_num_rows = (
              self._target_max_block_size // num_bytes_per_row
          )
          target_num_rows = max(1, target_num_rows)
  
          # TODO(swang): If the buffer is finalized, try to create even
          # blocks?
  
          if target_num_rows < block.num_rows():
              # Use copy=True to avoid holding the entire block in memory.
              block_to_yield = block.slice(
                  0, target_num_rows, copy=True
              )
              block_remainder = block.slice(
                  target_num_rows, block.num_rows(), copy=True
              )
  
      self._buffer = DelegatingBlockBuilder()
      if block_remainder is not None:
          self._buffer.add_block(block_remainder)
  
      self._returned_at_least_one_block = True
      return block_to_yield
  
file_path: ray/python/ray/data/_internal/output_buffer.py
imports: from typing import Any
  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
label_desc: Use image node_id 7 for calling the BlockOutputBuffer obj's underlying member method code with example usage: obj.next() and returns: block_to_yield