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

Code before:
    ) -> 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
        )
        self._cluster_autoscaler = create_cluster_autoscaler(
            self._topology,

Review comment (bug):
The logging statements here are outside the `if self._data_context.print_on_execution_start:` conditional block (lines 177-185), which means they will execute regardless of the print_on_execution_start setting. This is inconsistent with the other execution start logging above. Consider moving these logger.debug calls inside the conditional block to respect the print_on_execution_start setting.
```suggestion
                # 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)
```

Code after:
    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(
            self._data_context,
            self._dataset_id,
            self._topology,