class_docstr: A lazy execution plan for a Dataset.
class_docstr_tok: ["A","lazy","execution","plan","for","a","Dataset","."]
docstr: Create a cosmetic string representation of this execution plan.
  
  Returns:
      The string representation of this execution plan.
docstr_tok: ["Create","a","cosmetic","string","representation","of","this","execution","plan",".","Returns",":","The","string","representation","of","this","execution","plan","."]
code: def get_plan_as_string(self, classname: str) -> str:
      """Create a cosmetic string representation of this execution plan.
  
      Returns:
          The string representation of this execution plan.
      """
      # NOTE: this is used for Dataset.__repr__ to give a user-facing string
      # representation. Ideally ExecutionPlan.__repr__ should be replaced with this
      # method as well.
  
      # Do not force execution for schema, as this method is expected to be very
      # cheap.
      plan_str = ""
      plan_max_depth = 0
      dataset_blocks = None
      if (
          self._snapshot_blocks is None
          or self._snapshot_operator != self._logical_plan.dag
      ):
  
          def generate_logical_plan_string(
              op: LogicalOperator,
              curr_str: str = "",
              depth: int = 0,
          ):
              """Traverse (DFS) the LogicalPlan DAG and
              return a string representation of the operators."""
              if isinstance(op, (Read, InputData, AbstractFrom)):
                  return curr_str, depth
  
              curr_max_depth = depth
              op_name = op.name
              if depth == 0:
                  curr_str += f"{op_name}\n"
              else:
                  trailing_space = " " * ((depth - 1) * 3)
                  curr_str += f"{trailing_space}+- {op_name}\n"
  
              for input in op.input_dependencies:
                  (
                      curr_str,
                      input_max_depth,
                  ) = generate_logical_plan_string(
                      input, curr_str, depth + 1
                  )
                  curr_max_depth = max(curr_max_depth, input_max_depth)
              return curr_str, curr_max_depth
  
          # generate_logical_plan_string(self._logical_plan.dag)
          plan_str, plan_max_depth = generate_logical_plan_string(
              self._logical_plan.dag
          )
  
          # Get schema of initial blocks.
          if self.needs_eager_execution():
              # In the case where the plan contains only a Read/From operator,
              # it is cheap to execute it.
              # This allows us to get the most accurate estimates related
              # to the dataset, after applying execution plan optimizer rules
              # (e.g. number of blocks may change based on parallelism).
              self.execute()
          if self._snapshot_blocks is not None:
              schema = self._get_unified_blocks_schema(
                  self._snapshot_blocks, fetch_if_missing=False
              )
              dataset_blocks = self._snapshot_blocks
          else:
              assert self._in_blocks is not None
              schema = self._get_unified_blocks_schema(
                  self._in_blocks, fetch_if_missing=False
              )
              dataset_blocks = self._in_blocks
      else:
          # Get schema of output blocks.
          schema = self.schema(fetch_if_missing=False)
          dataset_blocks = self._snapshot_blocks
  
      if schema is None:
          schema_str = "Unknown schema"
      elif isinstance(schema, type):
          schema_str = str(schema)
      else:
          schema_str = []
          for n, t in zip(schema.names, schema.types):
              if hasattr(t, "__name__"):
                  t = t.__name__
              schema_str.append(f"{n}: {t}")
          schema_str = ", ".join(schema_str)
          schema_str = "{" + schema_str + "}"
      count = self._get_num_rows_from_blocks_metadata(dataset_blocks)
      if count is None:
          count = "?"
      if dataset_blocks is None:
          num_blocks = "?"
      else:
          num_blocks = dataset_blocks.estimated_num_blocks()
      name_str = (
          "name={}, ".format(self._dataset_name)
          if self._dataset_name is not None
          else ""
      )
      dataset_str = (
          "{}({}num_blocks={}, num_rows={}, schema={})".format(
              classname,
              name_str,
              num_blocks,
              count,
              schema_str,
          )
      )
  
      # If the resulting string representation fits in one line, use it directly.
      SCHEMA_LINE_CHAR_LIMIT = 80
      MIN_FIELD_LENGTH = 10
      INDENT_STR = " " * 3
      trailing_space = INDENT_STR * plan_max_depth
  
      if len(dataset_str) > SCHEMA_LINE_CHAR_LIMIT:
          # If the resulting string representation exceeds the line char limit,
          # first try breaking up each `Dataset` parameter into its own line
          # and check if each line fits within the line limit. We check the
          # `schema` param's length, since this is likely the longest string.
          schema_str_on_new_line = (
              f"{trailing_space}{INDENT_STR}schema={schema_str}"
          )
          if len(schema_str_on_new_line) > SCHEMA_LINE_CHAR_LIMIT:
              # If the schema cannot fit on a single line, break up each field
              # into its own line.
              schema_str = []
              for n, t in zip(schema.names, schema.types):
                  if hasattr(t, "__name__"):
                      t = t.__name__
                  col_str = f"{trailing_space}{INDENT_STR * 2}{n}: {t}"
                  # If the field line exceeds the char limit, abbreviate
                  # the field name to fit while maintaining the full type
                  if len(col_str) > SCHEMA_LINE_CHAR_LIMIT:
                      shortened_suffix = f"...: {str(t)}"
                      # Show at least 10 characters of the field name, even if
                      # we have already hit the line limit with the type.
                      chars_left_for_col_name = max(
                          SCHEMA_LINE_CHAR_LIMIT
                          - len(shortened_suffix),
                          MIN_FIELD_LENGTH,
                      )
                      col_str = f"{col_str[:chars_left_for_col_name]}{shortened_suffix}"
                  schema_str.append(col_str)
              schema_str = ",\n".join(schema_str)
              schema_str = (
                  "{\n"
                  + schema_str
                  + f"\n{trailing_space}{INDENT_STR}"
                  + "}"
              )
          name_str = (
              f"\n{trailing_space}{INDENT_STR}name={self._dataset_name},"
              if self._dataset_name is not None
              else ""
          )
          dataset_str = (
              f"{classname}("
              f"{name_str}"
              f"\n{trailing_space}{INDENT_STR}num_blocks={num_blocks},"
              f"\n{trailing_space}{INDENT_STR}num_rows={count},"
              f"\n{trailing_space}{INDENT_STR}schema={schema_str}"
              f"\n{trailing_space})"
          )
  
      if plan_max_depth == 0:
          plan_str += dataset_str
      else:
          plan_str += (
              f"{INDENT_STR * (plan_max_depth - 1)}+- {dataset_str}"
          )
      return plan_str
  
imports: import copy
  import itertools
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Union
  import ray
  from ray._private.internal_api import get_memory_info_reply, get_state_from_address
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.compute import ActorPoolStrategy, BlockTransform, CallableClass, ComputeStrategy, TaskPoolStrategy, UserDefinedFunction, get_compute, is_task_compute
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data._internal.logical.interfaces.logical_operator import LogicalOperator
  from ray.data._internal.logical.operators.from_operators import AbstractFrom
  from ray.data._internal.logical.operators.input_data_operator import InputData
  from ray.data._internal.logical.operators.read_operator import Read
  from ray.data._internal.logical.rules.operator_fusion import _are_remote_args_compatible
  from ray.data._internal.stats import DatasetStats, DatasetStatsSummary
  from ray.data._internal.util import create_dataset_tag, unify_block_metadata_schema
  from ray.data.block import Block, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.debug import log_once
label_desc: Use image node_id 3 for calling the ExecutionPlan obj's underlying member method code with example usage: obj.get_plan_as_string(classname) and returns: plan_str, curr_str,  curr_max_depth, curr_str,  depth