class_docstr: This rule eliminates unnecessary BuildOutputBlocksMapTransformFn,
  if the previous fn already outputs blocks.
  
  This happens for the "Read -> Map/Write" fusion.
code: def _optimize(
      self, transform_fns: List[MapTransformFn]
  ) -> List[MapTransformFn]:
      # For the following subsquence,
      # 1. Any MapTransformFn with block output.
      # 2. BuildOutputBlocksMapTransformFn
      # 3. Any MapTransformFn with block input.
      # We drop the BuildOutputBlocksMapTransformFn in the middle.
      new_transform_fns = []
  
      for i in range(len(transform_fns)):
          cur_fn = transform_fns[i]
          drop = False
          if (
              i > 0
              and i < len(transform_fns) - 1
              and isinstance(cur_fn, BuildOutputBlocksMapTransformFn)
          ):
              prev_fn = transform_fns[i - 1]
              next_fn = transform_fns[i + 1]
              if (
                  prev_fn.output_type == MapTransformFnDataType.Block
                  and next_fn.input_type == MapTransformFnDataType.Block
              ):
                  drop = True
          if not drop:
              new_transform_fns.append(cur_fn)
  
      return new_transform_fns
  
file_path: ray/python/ray/data/_internal/logical/rules/zero_copy_map_fusion.py
imports: from abc import abstractmethod
  from typing import List
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import BuildOutputBlocksMapTransformFn, MapTransformFn, MapTransformFnDataType
  from ray.data._internal.logical.interfaces.optimizer import Rule
  from ray.data._internal.logical.interfaces.physical_plan import PhysicalPlan
label_desc: Use image node_id 1 for calling the EliminateBuildOutputBlocks obj's underlying member method code with example usage: obj._optimize(transform_fns) and returns: new_transform_fns