Repository: ray-project/ray
File: python/ray/data/_internal/plan.py (Python)

Code before:
        self._dataset_uuid = None
        # Index of the current execution.
        self._run_index = -1

        self._dataset_name = None

        self._has_started_execution = False

        self._context = data_context

    def get_dataset_id(self) -> str:
        """Unique ID of the dataset, including the dataset name,
        UUID, and current execution index.
        """
        return (
            f"{self._dataset_name or 'dataset'}_{self._dataset_uuid}_{self._run_index}"
        )

    def create_executor(self) -> "StreamingExecutor":
        """Create an executor for this plan."""
        from ray.data._internal.execution.streaming_executor import StreamingExecutor

        self._run_index += 1
        callbacks = get_execution_callbacks(self._context)

        executor = StreamingExecutor(self._context, self.get_dataset_id(), callbacks)
        return executor

    def __repr__(self) -> str:
        return (
            f"ExecutionPlan("
            f"dataset_uuid={self._dataset_uuid}, "
            f"snapshot_operator={self._snapshot_operator}"
            f")"
        )

    def explain(self) -> str:
        """Return a string representation of the logical and physical plan."""

        convert_fns = [lambda x: x] + get_plan_conversion_fns()
        titles: List[str] = [
            "Logical Plan",
            "Logical Plan (Optimized)",
            "Physical Plan",
            "Physical Plan (Optimized)",
        ]

        # 1. Set initial plan
        plan = self._logical_plan

        sections = []

Review comment (suggestion):
Nit: Grouping callback construction with executor construction rather than incrementing the run index because they're more logically related
```suggestion
        
        callbacks = get_execution_callbacks(self._context)
        executor = StreamingExecutor(self._context, self.get_dataset_id(), callbacks)
```

Code after:
        )

    def create_executor(self) -> "StreamingExecutor":
        """Create an executor for this plan."""
        from ray.data._internal.execution.streaming_executor import StreamingExecutor

        self._run_index += 1
        executor = StreamingExecutor(self._context, self.get_dataset_id())
        return executor

    def __repr__(self) -> str:
        return (
            f"ExecutionPlan("
            f"dataset_uuid={self._dataset_uuid}, "
            f"snapshot_operator={self._cache._operator}"
            f")"
        )

    def explain(self) -> str:
        """Return a string representation of the logical and physical plan."""

        convert_fns = [lambda x: x] + get_plan_conversion_fns()
        titles: List[str] = [
            "Logical Plan",
            "Logical Plan (Optimized)",
            "Physical Plan",
            "Physical Plan (Optimized)",
        ]

        # 1. Set initial plan
        plan = self._logical_plan

        sections = []
        for title, convert_fn in zip(titles, convert_fns):

            # 2. Convert plan to new plan
            plan = convert_fn(plan)

            # 3. Generate plan str from new plan.
            plan_str, _ = self.generate_plan_string(plan.dag, show_op_repr=True)

            banner = f"\n-------- {title} --------\n"
            section = f"{banner}{plan_str}"
            sections.append(section)

        return "".join(sections)

    @staticmethod
    def generate_plan_string(
        op: Operator,
        curr_str: str = "",