class_docstr: Visitor optimizes the operation graph (see nodes.py).
  
  This operates on the translated graph which is emitted by the
  `_TranslateVisitor`, and performs optimizations.
  
  Namely, when enabled, this enables reading and writing from/to analyzer
  accumulator cache to avoid recomputing them over already seen datasets.
  This type of optimization requires also creating a partitioned view of the
  input data, according to the `is_partitionable` annotation.
code: def visit(self, operation_def, input_values):
      self._validate_operation_def(operation_def)
  
      if (
          isinstance(operation_def, beam_nodes.ApplySavedModel)
          and operation_def.phase == 0
      ):
          return self._visit_apply_savedmodel_operation(
              operation_def, input_values
          )
  
      # When self._cache_dict is None this means that we shouldn't do any cacheing
      # for this pipeline, and so there's no need to create any fine grained
      # views.
      if (
          self._cache_dict is not None
          and operation_def.is_partitionable
      ):
          return self._visit_partitionable_operation(
              operation_def, input_values
          )
  
      if input_values and any(
          v.fine_grained_view and v.prefer_fine_grained_view
          for v in input_values
      ):
          # We can 'flatten' the cached outputs of the parent operation since this
          # operation doesn't support partitioning.
          disaggregated_input_values = []
          for view in input_values:
              disaggregated_input_values.extend(
                  view.fine_grained_view.values()
              )
  
          # Each cache item should be a single ValueNode.
          assert all(
              isinstance(value, nodes.ValueNode)
              for value in disaggregated_input_values
          )
  
          next_inputs = nodes.apply_multi_output_operation(
              beam_nodes.Flatten,
              *disaggregated_input_values,
              label=f"FlattenCache[{operation_def.label}]",
          )
      else:
          # Parent operation output is not cacheable, therefore we can just use
          # a flattened view.
          next_inputs = tuple(v.flattened_view for v in input_values)
  
      flattened_view = nodes.OperationNode(
          operation_def, next_inputs
      ).outputs
  
      return tuple(
          _OptimizationView(  # pylint: disable=g-complex-comprehension
              prefer_fine_grained_view=False,
              flattened_view=flat,
              fine_grained_view=None,
              hashed_path=None,
          )
          for flat in flattened_view
      )
  
imports: import collections
  import dataclasses
  import hashlib
  from typing import Collection, Dict, Mapping, Optional, Tuple
  from typing import OrderedDict
  import tensorflow
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import common_types
  from tensorflow_transform import graph_tools
  from tensorflow_transform import impl_helper
  from tensorflow_transform import nodes
  from tensorflow_transform import tf2_utils
  from tensorflow_transform import tf_utils
  from tensorflow_transform.beam import analyzer_cache
  from tensorflow_transform.beam import beam_nodes
  from tensorflow_transform.beam import combiner_packing_util
label_desc: Use image node_id 5 for calling the _OptimizeVisitor obj's underlying member method code with example usage: obj.visit(operation_def, input_values) and returns: tuple, self, self