[data] refine backpressure info on progress bar (#51697)

## Why are these changes needed?

Refine backpressure info on progress bar to avoid confusion
* Add backpresure types
* Add remaining budgets for reading task outputs for the debug message.

Example:
```
Running Dataset. Active & requested resources: 2.1/12 CPU, 800.0MB/1.0GB object store: : 28.0 row [00:20, 1.26s/ row]
- ReadRange->Map(map1): Tasks: 1 [backpressured:tasks,outputs]; Queued blocks: 99; Resources: 1.0 CPU, 640.0MB object store (in=320.0MB,out=320.0MB), budget=(cpu=5.0,gpu=inf,obj_store=0.0B,out=0.0B): : 30.0 row [00:19, 1.44 row/s]
- Map(map2): Tasks: 1 [backpressured:tasks]; Queued blocks: 1; Resources: 1.1 CPU, 160.0MB object store (in=160.0MB,out=0.0B), budget=(cpu=4.9,gpu=inf,obj_store=96.0MB,out=224.0MB): : 26.0 row [00:19, 1.29 row/s]
```

## Related issue number

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


---------

Signed-off-by: Hao Chen <chenh1024@gmail.com>

Author: raulchen

python/ray/data/_internal/execution/resource_manager.py

@@ -250,7 +250,12 @@ def get_op_usage_str(self, op: PhysicalOperator) -> str:
                 budget = self._op_resource_allocator._op_budgets[op]
                 usage_str += f", budget=(cpu={budget.cpu:.1f}"
                 usage_str += f",gpu={budget.gpu:.1f}"
-                usage_str += f",object store={budget.object_store_memory_str()})"
+                usage_str += f",obj_store={budget.object_store_memory_str()}"
+                # Remaining memory budget for producing new task outputs.
+                reserved_for_output = memory_string(
+                    self._op_resource_allocator._output_budgets.get(op, 0)
+                )
+                usage_str += f",out={reserved_for_output})"
         return usage_str
 
     def op_resource_allocator_enabled(self) -> bool:
@@ -405,6 +410,8 @@ def __init__(self, resource_manager: ResourceManager, reservation_ratio: float):
         self._total_shared = ExecutionResources.zero()
         # Resource budgets for each operator, excluding `_reserved_for_op_outputs`.
         self._op_budgets: Dict[PhysicalOperator, ExecutionResources] = {}
+        # Remaining memory budget for generating new task outputs, per operator.
+        self._output_budgets: Dict[PhysicalOperator, float] = {}
         # Whether each operator has reserved the minimum resources to run
         # at least one task.
         # This is used to avoid edge cases where the entire resource limits are not
@@ -553,12 +560,14 @@ def max_task_output_bytes_to_read(self, op: PhysicalOperator) -> Optional[int]:
         op_outputs_usage = self._get_op_outputs_usage_with_downstream(op)
         res += max(self._reserved_for_op_outputs[op] - op_outputs_usage, 0)
         if math.isinf(res):
+            self._output_budgets[op] = res
             return None
 
         res = int(res)
         assert res >= 0
         if res == 0 and self._should_unblock_streaming_output_backpressure(op):
             res = 1
+        self._output_budgets[op] = res
         return res
 
     def _get_downstream_ineligible_ops(

python/ray/data/_internal/execution/streaming_executor_state.py

@@ -272,7 +272,14 @@ def summary_str(self, resource_manager: ResourceManager) -> str:
             self.op._in_task_submission_backpressure
             or self.op._in_task_output_backpressure
         ):
-            desc += " [backpressured]"
+            backpressure_types = []
+            if self.op._in_task_submission_backpressure:
+                # The op is backpressured from submitting new tasks.
+                backpressure_types.append("tasks")
+            if self.op._in_task_output_backpressure:
+                # The op is backpressured from producing new outputs.
+                backpressure_types.append("outputs")
+            desc += f" [backpressured:{','.join(backpressure_types)}]"
 
         # Actors info
         desc += self.op.actor_info_progress_str()