code: def _get_initial_stats_from_plan(plan: ExecutionPlan) -> DatasetStats:
      assert DataContext.get_current().optimizer_enabled
      if (
          plan._snapshot_blocks is not None
          and not plan._snapshot_blocks.is_cleared()
      ):
          return plan._snapshot_stats
      # For Datasets created from "read_xxx", `plan._in_blocks` is a LazyBlockList,
      # and `plan._in_stats` contains useless data.
      # For Datasets created from "from_xxx", we need to use `plan._in_stats` as
      # the initial stats. Because the `FromXxx` logical operators will be translated to
      # "InputDataBuffer" physical operators, which will be ignored when generating
      # stats, see `StreamingExecutor._generate_stats`.
      # TODO(hchen): Unify the logic by saving the initial stats in `InputDataBuffer
      if isinstance(plan._in_blocks, LazyBlockList):
          return DatasetStats(metadata={}, parent=None)
      else:
          return plan._in_stats
  
imports: from typing import Iterator, Tuple
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.execution.interfaces import Executor, PhysicalOperator, RefBundle
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data._internal.logical.interfaces.logical_plan import LogicalPlan
  from ray.data._internal.logical.operators.read_operator import Read
  from ray.data._internal.logical.optimizers import get_execution_plan
  from ray.data._internal.logical.rules.set_read_parallelism import compute_additional_split_factor
  from ray.data._internal.logical.util import record_operators_usage
  from ray.data._internal.plan import ExecutionPlan
  from ray.data._internal.planner.plan_read_op import apply_output_blocks_handling_to_read_task
  from ray.data._internal.stats import DatasetStats
  from ray.data.block import Block, BlockMetadata, List
  from ray.data.context import DataContext
  from ray.types import ObjectRef
label_desc: Use image node_id 6 for calling a global function with example usage: _get_initial_stats_from_plan(plan) and returns: plan, DatasetStats, plan