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

Code before:

    def execute(
        self, dag: PhysicalOperator, initial_stats: Optional[DatasetStats] = None
    ) -> OutputIterator:
        """Executes the DAG using a streaming execution strategy.

        We take an event-loop approach to scheduling. We block on the next scheduling
        event using `ray.wait`, updating operator state and dispatching new tasks.
        """

        self._initial_stats = initial_stats
        self._start_time = time.perf_counter()

        if not isinstance(dag, InputDataBuffer):
            if self._data_context.print_on_execution_start:
                message = f"Starting execution of Dataset {self._dataset_id}."
                log_path = get_log_directory()
                if log_path is not None:
                    message += f" Full logs are in {log_path}"
                logger.info(message)
                logger.info(
                    f"Execution plan of Dataset {self._dataset_id}: {dag.dag_str}"
                )

            # Log the full DataContext for traceability
            option_str = pprint.pformat(vars(self._options))
            logger.debug("Execution config:\n%s", option_str)
            context_str = pprint.pformat(vars(self._data_context))
            logger.debug("Data Context:\n%s", context_str)

        # Setup the streaming DAG topology and start the runner thread.
        self._topology = build_streaming_topology(dag, self._options)

        self._resource_manager = ResourceManager(
            self._topology,
            self._options,
            lambda: self._cluster_autoscaler.get_total_resources(),
            self._data_context,
        )

        # Setup progress manager
        self._progress_manager = get_progress_manager(
            self._data_context,
            self._dataset_id,
            self._topology,
            self._options.verbose_progress,
        )
        self._progress_manager.start()

        self._backpressure_policies = get_backpressure_policies(
            self._data_context, self._topology, self._resource_manager

Review comment (suggestion):
Using `vars(self._options)` bypasses the existing `__repr__` method defined in ExecutionOptions and exposes internal backing fields like `_resource_limits` instead of the public property `resource_limits`. This makes the log output less readable and inconsistent with the intended API. Consider using `str(self._options)` or `repr(self._options)` instead, which will use the well-defined __repr__ method that formats the output with public property names.
```suggestion
            option_str = repr(self._options)
```

Code after:
            tag_keys=("dataset", "operator"),
        )

    def execute(
        self,
        dag: PhysicalOperator,
        initial_stats: Optional[DatasetStats] = None,
        callbacks: Optional[List] = None,
    ) -> OutputIterator:
        """Executes the DAG using a streaming execution strategy.

        We take an event-loop approach to scheduling. We block on the next scheduling
        event using `ray.wait`, updating operator state and dispatching new tasks.
        """

        self._callbacks = callbacks if callbacks is not None else []
        self._initial_stats = initial_stats
        self._start_time = time.perf_counter()

        if not isinstance(dag, InputDataBuffer):
            if self._data_context.print_on_execution_start:
                message = f"Starting execution of Dataset {self._dataset_id}."
                log_path = get_log_directory()
                if log_path is not None:
                    message += f" Full logs are in {log_path}"
                logger.info(message)
                logger.info(
                    f"Execution plan of Dataset {self._dataset_id}: {dag.dag_str}"
                )

            # Log the full DataContext for traceability
            if logger.isEnabledFor(logging.DEBUG) and log_once(
                f"ray_data_log_context_{self._dataset_id}"
            ):
                logger.debug(
                    f"Data Context for dataset {self._dataset_id}:\n%s",
                    pprint.pformat(self._data_context),
                )

        # Setup the streaming DAG topology and start the runner thread.
        self._topology = build_streaming_topology(dag, self._options)

        self._resource_manager = ResourceManager(
            self._topology,
            self._options,
            lambda: self._cluster_autoscaler.get_total_resources(),
            self._data_context,
        )

        # Setup progress manager
        self._progress_manager = get_progress_manager(