[Data] Fix resource reservation by excluding completed operators' usages (#56319)

<!-- 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. -->

## Problem
The `ReservationOpResourceAllocator` was incorrectly accounting for
resource usage when calculating available resources for reservation.
Specifically, it wasn't properly handling completed operators who have
blocks in the output queue.

The `ReadFiles` operator below consumes 50 GB of object store memory and
should be excluded from reservation, but it is currently not.

<img width="1628" height="281" alt="image"
src="https://github.com/user-attachments/assets/8a80902d-7f88-4263-bc97-a3dee519b401"
/>

## Solution
Added logic to identify and subtract resource usage specifically from
completed physical operators:

## Testing results

Before the fix

<img width="958" height="653" alt="image"
src="https://github.com/user-attachments/assets/432dc94e-bbe1-4ecb-b1c3-a6e201da724a"
/>

After the fix

<img width="1567" height="700" alt="image"
src="https://github.com/user-attachments/assets/2b780d68-a208-4c6b-a250-dee0823e9083"
/>


## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] 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
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: xgui <xgui@anyscale.com>
Signed-off-by: Xinyuan <43737116+xinyuangui2@users.noreply.github.com>
Co-authored-by: Alexey Kudinkin <alexey.kudinkin@gmail.com>

Author: xinyuangui2

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

@@ -493,18 +493,56 @@ def _get_eligible_ops(self) -> List[PhysicalOperator]:
             op for op in self._resource_manager._topology if self._is_op_eligible(op)
         ]
 
+    def _get_ineligible_ops_with_usage(self) -> List[PhysicalOperator]:
+        """
+        Resource reservation is based on the number of eligible operators.
+        However, there might be completed operators that still have blocks in their output queue, which we need to exclude them from the reservation.
+        And we also need to exclude the downstream ineligible operators.
+
+        E.g., for the following pipeline:
+        ```
+        map1 (completed, but still has blocks in its output queue) -> limit1 (ineligible, not completed) -> map2 (not completed) -> limit2 -> map3
+        ```
+
+        The reservation is based on the number of eligible operators (map2 and map3), but we need to exclude map1 and limit1 from the reservation.
+        """
+        last_completed_ops = []
+        ops_to_exclude_from_reservation = []
+        # Traverse operator tree collecting all operators that have already finished
+        for op in self._resource_manager._topology:
+            if not op.execution_finished():
+                for dep in op.input_dependencies:
+                    if dep.execution_finished():
+                        last_completed_ops.append(dep)
+
+        # In addition to completed operators,
+        # filter out downstream ineligible operators since they are omitted from reservation calculations.
+        for op in last_completed_ops:
+            ops_to_exclude_from_reservation.extend(
+                list(self._get_downstream_ineligible_ops(op))
+            )
+            ops_to_exclude_from_reservation.append(op)
+        return list(set(ops_to_exclude_from_reservation))
+
     def _update_reservation(self):
-        global_limits = self._resource_manager.get_global_limits()
+        global_limits = self._resource_manager.get_global_limits().copy()
         eligible_ops = self._get_eligible_ops()
 
         self._op_reserved.clear()
         self._reserved_for_op_outputs.clear()
         self._reserved_min_resources.clear()
-        remaining = global_limits.copy()
 
         if len(eligible_ops) == 0:
             return
 
+        op_to_exclude_from_reservation = self._get_ineligible_ops_with_usage()
+        for completed_op in op_to_exclude_from_reservation:
+            global_limits = global_limits.subtract(
+                self._resource_manager.get_op_usage(completed_op)
+            )
+            global_limits = global_limits.max(ExecutionResources.zero())
+        remaining = global_limits.copy()
+
         # Reserve `reservation_ratio * global_limits / num_ops` resources for each
         # operator.
         default_reserved = global_limits.scale(