class_docstr: Physical operator for limit.
class_docstr_tok: ["Physical","operator","for","limit","."]
code: def _add_input_inner(self, refs: RefBundle, input_index: int) -> None:
      assert not self.completed()
      assert input_index == 0, input_index
      if self._limit_reached():
          return
      out_blocks: List[ObjectRef[Block]] = []
      out_metadata: List[BlockMetadata] = []
      for block, metadata in refs.blocks:
          num_rows = metadata.num_rows
          assert num_rows is not None
          if self._consumed_rows + num_rows <= self._limit:
              out_blocks.append(block)
              out_metadata.append(metadata)
              self._output_metadata.append(metadata)
              self._consumed_rows += num_rows
          else:
              # Slice the last block.
              def slice_fn(
                  block, metadata, num_rows
              ) -> Tuple[Block, BlockMetadata]:
                  block = BlockAccessor.for_block(block).slice(
                      0, num_rows, copy=True
                  )
                  metadata = copy.deepcopy(metadata)
                  metadata.num_rows = num_rows
                  metadata.size_bytes = BlockAccessor.for_block(
                      block
                  ).size_bytes()
                  return block, metadata
  
              block, metadata_ref = cached_remote_fn(
                  slice_fn, num_returns=2
              ).remote(
                  block,
                  metadata,
                  self._limit - self._consumed_rows,
              )
              out_blocks.append(block)
              metadata = ray.get(metadata_ref)
              out_metadata.append(metadata)
              self._output_metadata.append(metadata)
              self._consumed_rows = self._limit
              break
      self._cur_output_bundles += 1
      out_refs = RefBundle(
          list(zip(out_blocks, out_metadata)),
          owns_blocks=refs.owns_blocks,
      )
      self._buffer.append(out_refs)
      if self._limit_reached():
          self.all_inputs_done()
  
      # We cannot estimate if we have only consumed empty blocks
      if self._consumed_rows > 0:
          # Estimate number of output bundles
          # Check the case where _limit > # of input rows
          num_inputs = self.input_dependencies[0].num_outputs_total()
          estimated_total_output_rows = min(
              self._limit,
              self._consumed_rows
              / self._cur_output_bundles
              * num_inputs,
          )
          # _consumed_rows / _limit is roughly equal to
          # _cur_output_bundles / total output blocks
          self._estimated_output_blocks = round(
              estimated_total_output_rows
              / self._consumed_rows
              * self._cur_output_bundles
          )
  
imports: import copy
  from collections import deque
  from typing import Deque, List, Tuple
  import ray
  from ray.data._internal.execution.interfaces import PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import OneToOneOperator
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.stats import StatsDict
  from ray.data.block import Block, BlockAccessor, BlockMetadata
  from ray.types import ObjectRef
label_desc: Use image node_id 4 for calling the LimitOperator obj's underlying member method code with example usage: obj._add_input_inner(refs, input_index) and returns: block,  metadata