class_docstr: A lazy execution plan for a Dataset.
class_docstr_tok: ["A","lazy","execution","plan","for","a","Dataset","."]
docstr: Execute this plan.
  
  Args:
      allow_clear_input_blocks: Whether we should try to clear the input blocks
          for each stage.
      force_read: Whether to force the read stage to fully execute.
      preserve_order: Whether to preserve order in execution.
  
  Returns:
      The blocks of the output dataset.
docstr_tok: ["Execute","this","plan",".","Args",":","allow_clear_input_blocks",":","Whether","we","should","try","to","clear","the","input","blocks","for","each","stage",".","force_read",":","Whether","to","force","the","read","stage","to","fully","execute",".","preserve_order",":","Whether","to","preserve","order","in","execution",".","Returns",":","The","blocks","of","the","output","dataset","."]
code: def execute(
      self,
      allow_clear_input_blocks: bool = True,
      force_read: bool = False,
      preserve_order: bool = False,
  ) -> BlockList:
      """Execute this plan.
  
      Args:
          allow_clear_input_blocks: Whether we should try to clear the input blocks
              for each stage.
          force_read: Whether to force the read stage to fully execute.
          preserve_order: Whether to preserve order in execution.
  
      Returns:
          The blocks of the output dataset.
      """
  
      # Always used the saved context for execution.
      context = self._context
  
      if not ray.available_resources().get("CPU"):
          if log_once("cpu_warning"):
              logger.get_logger().warning(
                  "Warning: The Ray cluster currently does not have "
                  "any available CPUs. The Dataset job will hang unless more CPUs "
                  "are freed up. A common reason is that cluster resources are "
                  "used by Actors or Tune trials; see the following link "
                  "for more details: "
                  "https://docs.ray.io/en/latest/data/data-internals.html#ray-data-and-tune"  # noqa: E501
              )
      if not self.has_computed_output():
          from ray.data._internal.execution.legacy_compat import (
              _get_initial_stats_from_plan,
              execute_to_legacy_block_list,
              get_legacy_lazy_block_list_read_only,
          )
  
          if self.is_from_in_memory_only():
              # No need to execute MaterializedDatasets with only an InputData
              # operator, since the data is already materialized. This also avoids
              # recording unnecessary metrics for an empty plan execution.
              blocks = self._in_blocks
              stats = _get_initial_stats_from_plan(self)
          elif self.is_read_only():
              # If the Dataset is read-only, get the LazyBlockList without
              # executing the plan by only fetching metadata available from
              # the input Datasource or Reader without executing its ReadTasks.
              blocks = get_legacy_lazy_block_list_read_only(self)
              stats = _get_initial_stats_from_plan(self)
          else:
              from ray.data._internal.execution.streaming_executor import (
                  StreamingExecutor,
              )
  
              metrics_tag = create_dataset_tag(
                  self._dataset_name, self._dataset_uuid
              )
              executor = StreamingExecutor(
                  copy.deepcopy(context.execution_options),
                  metrics_tag,
              )
              blocks = execute_to_legacy_block_list(
                  executor,
                  self,
                  allow_clear_input_blocks=allow_clear_input_blocks,
                  dataset_uuid=self._dataset_uuid,
                  preserve_order=preserve_order,
              )
              stats = executor.get_stats()
              stats_summary_string = stats.to_summary().to_string(
                  include_parent=False
              )
              logger.get_logger(
                  log_to_stdout=context.enable_auto_log_stats
              ).info(
                  stats_summary_string,
              )
          # TODO(ekl) we shouldn't need to set this in the future once we move
          # to a fully lazy execution model, unless .materialize() is used. Th
          # reason we need it right now is since the user may iterate over a
          # Dataset multiple times after fully executing it once.
          if not self._run_by_consumer:
              blocks._owned_by_consumer = False
  
          # Retrieve memory-related stats from ray.
          reply = get_memory_info_reply(
              get_state_from_address(
                  ray.get_runtime_context().gcs_address
              )
          )
          if reply.store_stats.spill_time_total_s > 0:
              stats.global_bytes_spilled = int(
                  reply.store_stats.spilled_bytes_total
              )
          if reply.store_stats.restore_time_total_s > 0:
              stats.global_bytes_restored = int(
                  reply.store_stats.restored_bytes_total
              )
  
          stats.dataset_bytes_spilled = 0
  
          def collect_stats(cur_stats):
              stats.dataset_bytes_spilled += (
                  cur_stats.extra_metrics.get(
                      "obj_store_mem_spilled", 0
                  )
              )
              for parent in cur_stats.parents:
                  collect_stats(parent)
  
          collect_stats(stats)
  
          # Set the snapshot to the output of the final stage.
          self._snapshot_blocks = blocks
          self._snapshot_operator = self._logical_plan.dag
          self._snapshot_stats = stats
          self._snapshot_stats.dataset_uuid = self._dataset_uuid
  
          # In the case of a read-only dataset, we replace the
          # input LazyBlockList with a copy that includes the
          # calculated metadata from initializing the InputDataBuffer.
          if self.is_read_only():
              self._in_blocks = blocks
      if _is_lazy(self._snapshot_blocks) and force_read:
          executed_blocks = self._snapshot_blocks.compute_to_blocklist()
          # After executing the snapshot blocks, get its updated stats.
          # The snapshot blocks after execution will contain the execution stats.
          self._snapshot_stats = self._snapshot_blocks.stats()
          self._snapshot_blocks = executed_blocks
          self._snapshot_operator = self._logical_plan.dag
          # When force-read is enabled, we similarly update self._in_blocks.
          if self.is_read_only():
              self._in_blocks = self._snapshot_blocks
      return self._snapshot_blocks
  
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 15 for calling the ExecutionPlan obj's underlying member method code with example usage: obj.execute(allow_clear_input_blocks, force_read, preserve_order) and returns: self