[Data] Return 'inf' for max object store memory usage (ray-project#59751)

ray-project#59412 updated the reservation
resource allocator to cap resource allocations based on the resources
returned by `min_max_resource_usage`. The problem is that running tasks
can produce any amounts of data, so it doesn't make sense to cap by
`obj_store_mem_max_pending_output_per_task * concurrency`.

This PR fixes that issue by setting the max object store memory usage to
'inf'.

Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu>
Signed-off-by: jasonwrwang <jasonwrwang@tencent.com>

Author: bveeramani

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

@@ -484,7 +484,9 @@ def min_max_resource_requirements(
                 cpu=num_cpus_per_actor * max_actors,
                 gpu=num_gpus_per_actor * max_actors,
                 memory=memory_per_actor * max_actors,
-                object_store_memory=obj_store_mem_per_task * max_actors,
+                # Set the max `object_store_memory` requirement to 'inf', because we
+                # don't know how much data each task can output.
+                object_store_memory=float("inf"),
             )
 
         return min_resource_usage, max_resource_usage

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

@@ -188,7 +188,9 @@ def min_max_resource_requirements(
                 cpu=per_task.cpu * self._max_concurrency,
                 gpu=per_task.gpu * self._max_concurrency,
                 memory=per_task.memory * self._max_concurrency,
-                object_store_memory=obj_store_per_task * self._max_concurrency,
+                # Set the max `object_store_memory` requirement to 'inf', because we
+                # don't know how much data each task can output.
+                object_store_memory=float("inf"),
             )
         else:
             max_resource_usage = ExecutionResources.for_limits()

python/ray/data/tests/test_actor_pool_map_operator.py

@@ -680,9 +680,11 @@ def test_min_max_resource_requirements(restore_data_context):
     ) = op.min_max_resource_requirements()
 
     # min_resource_usage: 1 actor * (1 gpu, 3 obj_store_mem)
-    # max_resource_usage: 2 actors * (1 gpu, 3 obj_store_mem)
+    # max_resource_usage: 2 actors * (1 gpu)
     assert min_resource_usage_bound == ExecutionResources(gpu=1, object_store_memory=3)
-    assert max_resource_usage_bound == ExecutionResources(gpu=2, object_store_memory=6)
+    assert max_resource_usage_bound == ExecutionResources(
+        gpu=2, object_store_memory=float("inf")
+    )
 
 
 def test_min_max_resource_requirements_unbounded(restore_data_context):