class_docstr: Chunks blocks into batches.
docstr: Get the next batch from the block buffer.
  
  Returns:
      A batch represented as a Block.
docstr_tok: ["Get","the","next","batch","from","the","block","buffer",".","Returns",":","A","batch","represented","as","a","Block","."]
code: def next_batch(self) -> Block:
      """Get the next batch from the block buffer.
  
      Returns:
          A batch represented as a Block.
      """
      assert self.has_batch() or (self._done_adding and self.has_any())
      needs_copy = self._ensure_copy
      # If no batch size, short-circuit.
      if self._batch_size is None:
          assert len(self._buffer) == 1
          block = self._buffer[0]
          if needs_copy:
              # Copy block if needing to ensure fresh batch copy.
              block = BlockAccessor.for_block(block)
              block = block.slice(0, block.num_rows(), copy=True)
          self._buffer = []
          self._buffer_size = 0
          return block
      output = DelegatingBlockBuilder()
      leftover = []
      needed = self._batch_size
      for block in self._buffer:
          accessor = BlockAccessor.for_block(block)
          if needed <= 0:
              # We already have a full batch, so add this block to
              # the leftovers.
              leftover.append(block)
          elif accessor.num_rows() <= needed:
              # We need this entire block to fill out a batch.
              # We need to call `accessor.slice()` to ensure
              # the subsequent block's type are the same.
              output.add_block(
                  accessor.slice(0, accessor.num_rows(), copy=False)
              )
              needed -= accessor.num_rows()
          else:
              if (
                  isinstance(accessor, ArrowBlockAccessor)
                  and block.num_columns > 0
                  and block.column(0).num_chunks
                  >= MIN_NUM_CHUNKS_TO_TRIGGER_COMBINE_CHUNKS
              ):
                  accessor = BlockAccessor.for_block(
                      transform_pyarrow.combine_chunks(block)
                  )
              # We only need part of the block to fill out a batch.
              output.add_block(accessor.slice(0, needed, copy=False))
              # Add the rest of the block to the leftovers.
              leftover.append(
                  accessor.slice(
                      needed, accessor.num_rows(), copy=False
                  )
              )
              needed = 0
  
      # Move the leftovers into the block buffer so they're the first
      # blocks consumed on the next batch extraction.
      self._buffer = leftover
      self._buffer_size -= self._batch_size
      needs_copy = needs_copy and not output.will_build_yield_copy()
      batch = output.build()
      if needs_copy:
          # Need to ensure that the batch is a fresh copy.
          batch = BlockAccessor.for_block(batch)
          batch = batch.slice(0, batch.num_rows(), copy=True)
      return batch
  
imports: from typing import Optional
  from ray.data._internal.arrow_block import ArrowBlockAccessor
  from ray.data._internal.arrow_ops import transform_pyarrow
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data.block import Block, BlockAccessor
label_desc: Use image node_id 6 for calling the Batcher obj's underlying member method code with example usage: obj.next_batch() and returns: batch, block