code: def apply_to_operand(self, op: OperandType):
      node = op.outputs[0]
      input_node = self._graph.predecessors(node)[0]
      nrows = input_node.op.nrows or 0
      head = op.indexes[0].stop
  
      new_input_op = input_node.op.copy()
      new_input_op._key = input_node.op.key
      new_input_op.nrows = nrows = max(nrows, head)
      new_input_params = input_node.params.copy()
      new_input_params["shape"] = (nrows,) + input_node.shape[1:]
      pandas_index = node.index_value.to_pandas()[:nrows]
      new_input_params["index_value"] = parse_index(pandas_index, node)
      new_input_params.update(input_node.extra_params)
      new_entity = (
          new_input_op.new_tileable
          if not isinstance(node, CHUNK_TYPE)
          else new_input_op.new_chunk
      )
      new_input_node = new_entity(
          input_node.inputs, kws=[new_input_params]
      ).data
  
      if (
          new_input_node.op.nrows == head
          and self._graph.count_successors(input_node) == 1
      ):
          new_input_node._key = node.key
          new_input_node._id = node.id
          # just remove the input data
          self._graph.add_node(new_input_node)
          for succ in self._graph.successors(node):
              self._graph.add_edge(new_input_node, succ)
          for pred in self._graph.predecessors(input_node):
              self._graph.add_edge(pred, new_input_node)
          self._graph.remove_node(input_node)
          self._graph.remove_node(node)
  
          # mark optimization record
          # the input node is removed
          self._records.append_record(
              OptimizationRecord(
                  input_node, None, OptimizationRecordType.delete
              )
          )
          self._records.append_record(
              OptimizationRecord(
                  node, new_input_node, OptimizationRecordType.replace
              )
          )
          new_node = new_input_node
      else:
          self._replace_node(input_node, new_input_node)
          new_op = op.copy()
          new_op._key = op.key
          params = node.params.copy()
          params.update(node.extra_params)
          new_entity = (
              new_op.new_tileable
              if not isinstance(node, CHUNK_TYPE)
              else new_op.new_chunk
          )
          new_node = new_entity([new_input_node], kws=[params]).data
          self._replace_node(node, new_node)
  
          # mark optimization record
          self._records.append_record(
              OptimizationRecord(
                  input_node,
                  new_input_node,
                  OptimizationRecordType.replace,
              )
          )
          self._records.append_record(
              OptimizationRecord(
                  node, new_node, OptimizationRecordType.replace
              )
          )
  
      # check node if it's in result
      try:
          i = self._graph.results.index(node)
          self._graph.results[i] = new_node
      except ValueError:
          pass
  
file_path: mars/mars/optimization/logical/common/head.py
imports: from typing import List
  from ....core import OperandType, TileableType, CHUNK_TYPE
  from ....dataframe.base.value_counts import DataFrameValueCounts
  from ....dataframe.datasource.core import HeadOptimizedDataSource
  from ....dataframe.sort.core import DataFrameSortOperand
  from ....dataframe.utils import parse_index
  from ....utils import implements
  from ..core import OperandBasedOptimizationRule, OptimizationRecord, OptimizationRecordType
label_desc: Use image node_id 4 for calling the HeadPushDown obj's underlying member method code with example usage: obj.apply_to_operand(op) without return types