code: def apply_to_operand(self, op: OperandType):
      node = op.outputs[0]
      data_source_node = self._graph.predecessors(node)[0]
  
      if (
          isinstance(node, CHUNK_TYPE)
          and self._graph.count_successors(data_source_node) == 1
      ):
          # merge into data source only for chunk
          data_source_params = node.params.copy()
          data_source_params.update(data_source_node.extra_params)
          data_source_op = data_source_node.op.copy()
          data_source_op._key = data_source_node.op.key
          data_source_op._output_types = op.output_types
          if node.ndim == 1:
              data_source_op.set_pruned_columns(
                  node.name, keep_order=True
              )
          else:
              data_source_op.set_pruned_columns(
                  node.dtypes.index.tolist(), keep_order=True
              )
          new_entity = (
              data_source_op.new_tileable
              if not isinstance(node, CHUNK_TYPE)
              else data_source_op.new_chunk
          )
          new_data_source_node = new_entity(
              data_source_node.inputs, kws=[data_source_params]
          ).data
          new_data_source_node._key = node.key
          new_data_source_node._id = node.id
          # just remove the input data
          self._graph.add_node(new_data_source_node)
          for succ in self._graph.successors(node):
              self._graph.add_edge(new_data_source_node, succ)
          self._graph.remove_node(data_source_node)
          self._graph.remove_node(node)
  
          # mark optimization record
          # the input node is removed
          self._records.append_record(
              OptimizationRecord(
                  data_source_node, None, OptimizationRecordType.delete
              )
          )
          self._records.append_record(
              OptimizationRecord(
                  node,
                  new_data_source_node,
                  OptimizationRecordType.replace,
              )
          )
          new_outputs = [new_data_source_node]
      else:
          selected_columns: List[Any] = self._get_selected_columns(op)
          original_node = self._records.get_original_entity(
              data_source_node
          )
          if original_node is not None:
              # pruned before
              dtypes = original_node.dtypes
          else:
              dtypes = data_source_node.dtypes
          data_source_params = data_source_node.params.copy()
          data_source_params["shape"] = (
              data_source_node.shape[0],
              len(selected_columns),
          )
          data_source_params["dtypes"] = dtypes = dtypes[
              selected_columns
          ]
          data_source_params["columns_value"] = parse_index(
              dtypes.index, store_data=True
          )
          data_source_params.update(data_source_node.extra_params)
          data_source_node_op = data_source_node.op.copy()
          data_source_node_op._key = data_source_node.op.key
          data_source_node_op.set_pruned_columns(
              selected_columns, keep_order=True
          )
          new_data_source_node = data_source_node_op.new_tileable(
              data_source_node_op.inputs, kws=[data_source_params]
          ).data
  
          self._replace_node(data_source_node, new_data_source_node)
          # mark optimization record
          self._records.append_record(
              OptimizationRecord(
                  data_source_node,
                  new_data_source_node,
                  OptimizationRecordType.replace,
              )
          )
  
          new_op = op.copy()
          new_op._key = op.key
          kws = []
          for out in op.outputs:
              params = out.params.copy()
              params.update(out.extra_params)
              kws.append(params)
          new_entity = (
              new_op.new_tileables
              if not isinstance(node, CHUNK_TYPE)
              else new_op.new_chunks
          )
          new_outputs = [
              t.data
              for t in new_entity([new_data_source_node], kws=kws)
          ]
  
          for out, new_out in zip(op.outputs, new_outputs):
              new_out._id = out.id
              new_out._key = out.key
              self._replace_node(out, new_out)
              # mark optimization record
              self._records.append_record(
                  OptimizationRecord(
                      out, new_out, OptimizationRecordType.replace
                  )
              )
  
      for out, new_out in zip(op.outputs, new_outputs):
          # check out if it's in result
          try:
              i = self._graph.results.index(out)
              self._graph.results[i] = new_out
          except ValueError:
              pass
  
file_path: mars/mars/optimization/logical/common/column_pruning.py
imports: from abc import ABCMeta, abstractmethod
  from typing import Any, List
  from ..core import OperandBasedOptimizationRule, OptimizationRecord, OptimizationRecordType
  from ....core import OperandType, TileableType, CHUNK_TYPE
  from ....dataframe.datasource.core import ColumnPruneSupportedDataSourceMixin
  from ....dataframe.utils import parse_index
  from ....utils import implements
label_desc: Use image node_id 6 for calling the PruneDataSource obj's underlying member method code with example usage: obj.apply_to_operand(op) without return types