[data] Improve execution progress rendering (ray-project#56992)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?
<img width="1263" height="859" alt="Screenshot 2025-10-07 at 9 46 09 PM"
src="https://github.com/user-attachments/assets/45249e77-af49-4e3d-a758-608a51d15e10"
/>

[Link to video
recording](https://drive.google.com/file/d/13Erjd_K4OXmn1r_7iMS97u8cgUdBgTvG/view?usp=sharing)

Currently, by default, the original `tqdm` based progress is used. To
enable `rich` progress reporting as shown in the screenshot, set:
```
ray.data.DataContext.get_current().enable_rich_progress_bars = True
```
or set the envvar:
```
export RAY_DATA_ENABLE_RICH_PROGRESS_BARS=1
```

<!-- Please give a short summary of the change and the problem this
solves. -->

## Related issue number
Fixes ray-project#52505

<!-- For example: "Closes ray-project#1234" -->

## Checks

- [x] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [x] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Replaces legacy per-operator/global progress bars with a Rich-based
progress manager that tracks global/operator progress and resources,
refactors topology/progress APIs, and updates tests.
> 
> - **Execution Progress (Rich-based)**:
> - Introduces `progress_manager.py` with `RichExecutionProgressManager`
for global/operator progress, rates, elapsed/remaining time, and live
resource usage.
> - Streaming executor integrates manager (start/refresh/close,
finishing messages), updates on row/output and resources, and periodic
refresh via `PROGRESS_MANAGER_UPDATE_INTERVAL`.
> - **Operator/State Refactor**:
> - `OpState` gains `OpDisplayMetrics`, `progress_manager_uuid`,
`output_row_count`, and `update_display_metrics`; removes legacy
progress bar handling and summary methods.
> - `_debug_dump_topology` now logs `op_display_metrics.display_str()`.
> - Minor: add TODOs on sub-progress-bar helpers in `AllToAllOperator`
and `HashShuffleProgressBarMixin`.
> - **Topology API**:
> - `build_streaming_topology(...)` now returns only `Topology` (no
progress bar count); all call sites and tests updated.
> - **Iterator/Reporting**:
>   - `_ClosingIterator` updates total progress via manager.
>   - Resource reporting moved to progress manager.
> - **Tests**:
> - Adjust unit tests to new topology return type and removed progress
bar expectations.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
935f3c3. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Signed-off-by: Daniel Shin <kyuseung1016@gmail.com>
Signed-off-by: kyuds <kyuseung1016@gmail.com>
Signed-off-by: kyuds <kyuds@everspin.co.kr>

Author: kyuds

doc/source/data/monitoring-your-workload.rst

@@ -40,6 +40,9 @@ For operator names longer than a threshold of 100 characters, Ray Data truncates
 * To turn off this behavior and show the full operator name, set `DataContext.get_current().enable_progress_bar_name_truncation = False`.
 * To change the threshold of truncating the name, update the constant `ray.data._internal.progress_bar.ProgressBar.MAX_NAME_LENGTH = 42`.
 
+.. tip::
+   There is a new experimental console UI to show progress bars. Set `DataContext.get_current().enable_rich_progress_bars = True` or set the `RAY_DATA_ENABLE_RICH_PROGRESS_BARS=1` environment variable to enable.
+
 .. _ray-data-dashboard:
 
 Ray Data dashboard

python/ray/data/_internal/execution/operators/base_physical_operator.py

@@ -1,4 +1,4 @@
-import abc
+from abc import ABC, abstractmethod
 from typing import List, Optional
 
 from ray.data._internal.execution.interfaces import (
@@ -8,13 +8,14 @@
     TaskContext,
 )
 from ray.data._internal.execution.interfaces.physical_operator import _create_sub_pb
+from ray.data._internal.execution.operators.sub_progress import SubProgressBarMixin
 from ray.data._internal.logical.interfaces import LogicalOperator
 from ray.data._internal.stats import StatsDict
 from ray.data.context import DataContext
 
 
-class InternalQueueOperatorMixin(PhysicalOperator, abc.ABC):
-    @abc.abstractmethod
+class InternalQueueOperatorMixin(PhysicalOperator, ABC):
+    @abstractmethod
     def internal_queue_size(self) -> int:
         """Returns Operator's internal queue size"""
         ...
@@ -47,7 +48,9 @@ def input_dependency(self) -> PhysicalOperator:
         return self.input_dependencies[0]
 
 
-class AllToAllOperator(InternalQueueOperatorMixin, PhysicalOperator):
+class AllToAllOperator(
+    InternalQueueOperatorMixin, SubProgressBarMixin, PhysicalOperator
+):
     """A blocking operator that executes once its inputs are complete.
 
     This operator implements distributed sort / shuffle operations, etc.
@@ -169,6 +172,15 @@ def close_sub_progress_bars(self):
             for sub_bar in self._sub_progress_bar_dict.values():
                 sub_bar.close()
 
+    def get_sub_progress_bar_names(self) -> Optional[List[str]]:
+        return self._sub_progress_bar_names
+
+    def set_sub_progress_bar(self, name, pg):
+        # not type-checking due to circular imports
+        if self._sub_progress_bar_dict is None:
+            self._sub_progress_bar_dict = {}
+        self._sub_progress_bar_dict[name] = pg
+
     def supports_fusion(self):
         return True
 

python/ray/data/_internal/execution/operators/hash_shuffle.py

@@ -45,6 +45,7 @@
     _create_sub_pb,
     estimate_total_num_of_blocks,
 )
+from ray.data._internal.execution.operators.sub_progress import SubProgressBarMixin
 from ray.data._internal.logical.interfaces import LogicalOperator
 from ray.data._internal.stats import OpRuntimeMetrics
 from ray.data._internal.table_block import TableBlockAccessor
@@ -363,7 +364,7 @@ def combine(one: "_PartitionStats", other: "_PartitionStats") -> "_PartitionStat
         )
 
 
-class HashShuffleProgressBarMixin(abc.ABC):
+class HashShuffleProgressBarMixin(SubProgressBarMixin):
     @property
     @abc.abstractmethod
     def shuffle_name(self) -> str:
@@ -374,26 +375,29 @@ def shuffle_name(self) -> str:
     def reduce_name(self) -> str:
         ...
 
+    def _validate_sub_progress_bar_names(self):
+        assert self.shuffle_name is not None, "shuffle_name should not be None"
+        assert self.reduce_name is not None, "reduce_name should not be None"
+
     def initialize_sub_progress_bars(self, position: int) -> int:
-        """Display all sub progres bars in the termainl, and return the number of bars."""
+        """Display all sub progress bars in the termainl, and return the number of bars."""
+        self._validate_sub_progress_bar_names()
 
         # shuffle
         progress_bars_created = 0
         self.shuffle_bar = None
-        if self.shuffle_name is not None:
-            self.shuffle_bar, position = _create_sub_pb(
-                self.shuffle_name, self.num_output_rows_total(), position
-            )
-            progress_bars_created += 1
+        self.shuffle_bar, position = _create_sub_pb(
+            self.shuffle_name, self.num_output_rows_total(), position
+        )
+        progress_bars_created += 1
         self.shuffle_metrics = OpRuntimeMetrics(self)
 
         # reduce
         self.reduce_bar = None
-        if self.reduce_name is not None:
-            self.reduce_bar, position = _create_sub_pb(
-                self.reduce_name, self.num_output_rows_total(), position
-            )
-            progress_bars_created += 1
+        self.reduce_bar, position = _create_sub_pb(
+            self.reduce_name, self.num_output_rows_total(), position
+        )
+        progress_bars_created += 1
         self.reduce_metrics = OpRuntimeMetrics(self)
 
         return progress_bars_created
@@ -403,6 +407,27 @@ def close_sub_progress_bars(self):
         self.shuffle_bar.close()
         self.reduce_bar.close()
 
+    def get_sub_progress_bar_names(self) -> Optional[List[str]]:
+        self._validate_sub_progress_bar_names()
+
+        # shuffle
+        self.shuffle_bar = None
+        self.shuffle_metrics = OpRuntimeMetrics(self)
+
+        # reduce
+        self.reduce_bar = None
+        self.reduce_metrics = OpRuntimeMetrics(self)
+
+        return [self.shuffle_name, self.reduce_name]
+
+    def set_sub_progress_bar(self, name, pg):
+        # No type-hints due to circular imports. `name` should be a `str`
+        # and `pg` should be a `SubProgressBar`
+        if self.shuffle_name is not None and self.shuffle_name == name:
+            self.shuffle_bar = pg
+        elif self.reduce_name is not None and self.reduce_name == name:
+            self.reduce_bar = pg
+
 
 def _derive_max_shuffle_aggregators(
     total_cluster_resources: ExecutionResources,
@@ -705,7 +730,7 @@ def _on_partitioning_done(cur_shuffle_task_idx: int):
                 self.shuffle_metrics.on_task_finished(cur_shuffle_task_idx, None)
 
                 # Update Shuffle progress bar
-                self.shuffle_bar.update(i=input_block_metadata.num_rows)
+                self.shuffle_bar.update(increment=input_block_metadata.num_rows or 0)
 
             # TODO update metrics
             task = self._shuffling_tasks[input_index][
@@ -823,7 +848,7 @@ def _on_bundle_ready(partition_id: int, bundle: RefBundle):
 
             # Update Finalize progress bar
             self.reduce_bar.update(
-                i=bundle.num_rows(), total=self.num_output_rows_total()
+                increment=bundle.num_rows() or 0, total=self.num_output_rows_total()
             )
 
         def _on_aggregation_done(partition_id: int, exc: Optional[Exception]):

python/ray/data/_internal/execution/operators/sub_progress.py

@@ -0,0 +1,28 @@
+from abc import ABC, abstractmethod
+from typing import List, Optional
+
+
+class SubProgressBarMixin(ABC):
+    """Abstract class for operators that support sub-progress bars"""
+
+    @abstractmethod
+    def get_sub_progress_bar_names(self) -> Optional[List[str]]:
+        """
+        Returns list of sub-progress bar names
+
+        This is used to create the sub-progress bars in the progress manager.
+        Note that sub-progress bars will be created in the order returned by
+        this method.
+        """
+        ...
+
+    @abstractmethod
+    def set_sub_progress_bar(self, name, pg):
+        """
+        Sets sub-progress bars
+
+        name: name of sub-progress bar
+        pg: SubProgressBar instance (progress_manager.py)
+        """
+        # Skipping type-checking for circular imports
+        ...