class_docstr: Do additional splits on output blocks.
class_docstr_tok: ["Do","additional","splits","on","output","blocks","."]
code: def __call__(
      self, blocks: Iterable[Block], ctx: TaskContext
  ) -> Iterable[Block]:
      for block in blocks:
          block = BlockAccessor.for_block(block)
          offset = 0
          split_sizes = _splitrange(
              block.num_rows(), self._additional_split_factor
          )
          for size in split_sizes:
              # NOTE: copy=True is needed because this is an output block. If
              # a block slice is put into the object store, the entire block
              # will get serialized.
              yield block.slice(offset, offset + size, copy=True)
              offset += size
  
imports: import itertools
  from abc import abstractmethod
  from enum import Enum
  from typing import Any, Callable, Dict, Iterable, List, Optional, TypeVar, Union
  from ray.data._internal.block_batching.block_batching import batch_blocks
  from ray.data._internal.execution.interfaces.task_context import TaskContext
  from ray.data._internal.output_buffer import BlockOutputBuffer
  from ray.data.block import Block, BlockAccessor, DataBatch
label_desc: Use image node_id 2 for calling the ApplyAdditionalSplitToOutputBlocks obj's underlying member method code with example usage: obj.__call__(blocks, ctx) without return types