[Data] Decouple sub-progress metrics from progress bar implementations

[400Ping]

Summary

This PR refactors Ray Data sub-progress reporting so operators expose sub-progress as metrics instead of receiving concrete progress bar objects from progress managers.

Before this change, the flow looked like this:

1. Operator exposed sub-progress bar names.
2. Progress manager created sub-progress bar objects.
3. Progress manager injected those bar objects back into the operator.
4. Operator and scheduler logic updated those objects directly.

After this change, the flow is:

1. Operator exposes Dict[str, ProgressMetrics].
2. Driver-side execution updates sub-progress through internal updater helpers.
3. Progress managers read the metrics and attach display-specific state as needed.
4. Progress managers own rendering, while operators own execution state.

This removes the previous bi-directional coupling between operators and progress bar implementations.

What Changed

API changes

• Replace the sub-progress name-only interface with a metrics-based interface:
  • get_sub_progress_metrics() -> Dict[str, ProgressMetrics]
• ProgressMetrics is now a frozen dataclass snapshot with:
  • name
  • total
  • completed

Internal execution changes

• Introduce internal SubProgressUpdater helpers for driver-side mutation paths that still need:
  • update(...)
  • fetch_until_complete(...)
  • block_until_complete(...)
• TaskContext now carries:
  • sub_progress_metrics
  • sub_progress_updaters

Progress manager changes

• tqdm, rich, logging, and noop progress managers no longer inject sub-progress bar objects back into operators.
• Managers now attach display-specific state to internal updater helpers and render from metrics.
• NoopSubProgressBar is no longer needed.

Operator and scheduler updates

Updated the following sub-progress users to the new model:

• AllToAllOperator
• CPU hash shuffle
• GPU shuffle
• sort / aggregate sampling
• pull-based shuffle scheduler
• push-based shuffle scheduler
• split repartition scheduler

Why This Design

This PR keeps the external sub-progress interface simple and metrics-based, while preserving existing driver-side execution behavior for shuffle/sample paths that still need incremental wait-and-update helpers.

Conceptually:

Before:
Operator <-> ProgressManager <-> ProgressBar object

After:
Operator -> ProgressMetrics
Execution -> SubProgressUpdater -> ProgressMetrics
ProgressManager -> render ProgressMetrics

This is a middle ground that removes bar injection and keeps the metrics interface clean, without rewriting all driver-side progress update paths at once.

Related Issues

• Closes [Data] Decouple SubProgressMixin from progress bar #60082
• Related: [Data] Encapsulate isatty progress handling and decouple from TQDM progress bar #60083
• Related: [Data] Streaming executor logging verbosity not configurable? #42191
• Related: [Datasets] Dataset.map_batches progress bar is incorrect #33517

[gemini-code-assist]

Code Review

This pull request refactors the sub-progress tracking system in Ray Data, transitioning from a direct progress bar dictionary to a more robust metrics-based approach using ProgressMetrics and SubProgressUpdater. The changes involve updating various physical operators (AllToAll, HashShuffle, GPUShuffle) and progress managers (Logging, Rich, Tqdm) to utilize the new SubProgressMixin. Key feedback includes addressing potential type mismatches and state inconsistencies in legacy setters, ensuring thread safety in the SubProgressUpdater.update method to prevent race conditions, and improving maintainability by using constants for hardcoded strings and name lengths.

[github-actions]

This pull request has been automatically marked as stale because it has not had
any activity for 14 days. It will be closed in another 14 days if no further activity occurs.
Thank you for your contributions.

You can always ask for help on our discussion forum or Ray's public slack channel.

If you'd like to keep this open, just leave any comment, and the stale label will be removed.

[400Ping]

Unstale

[kyuds]

sorry for the delay @400Ping, but left some initial comments. If you could also rebase to latest master and resolve the merge conflicts, I'll do more testing on my side!

[kyuds]

whats the purpose of keeping this class frozen?

[400Ping]

It is frozen because ProgressMetrics is meant to be an immutable snapshot, not a mutable progress bar object. SubProgressUpdater updates progress by replacing the snapshot, while progress managers only read/sync from it. This keeps rendering code from mutating operator-owned execution state directly.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{Reviewed by Cursor Bugbot for commit f164692. Configure here.}

[400Ping]

Opening a new on at #63195 , there seems to be a PR git ref error