[data] set default memory limit fraction to 25% when memory reservati…

…on not enabled (#43686)

#43171 increased the default memory limit fraction to 50%. because with memory reservation, we have more precise control over the memory. 
This PR set the default back to 25%, when memory reservation is not enabled, to prevent regression. 

Signed-off-by: Hao Chen <[email protected]>

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

@@ -36,9 +36,15 @@ class ResourceManager:
     GLOBAL_LIMITS_UPDATE_INTERVAL_S = 10
 
     # The fraction of the object store capacity that will be used as the default object
-    # store memory limit for the streaming executor.
+    # store memory limit for the streaming executor,
+    # when `ReservationOpResourceAllocator` is enabled.
     DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION = 0.5
 
+    # The fraction of the object store capacity that will be used as the default object
+    # store memory limit for the streaming executor,
+    # when `ReservationOpResourceAllocator` is not enabled.
+    DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION_WO_RESOURCE_RESERVATION = 0.25
+
     # Memory accounting is accurate only for these operators.
     # We'll enable memory reservation if a dataset only contains these operators.
     _ACCURRATE_MEMORY_ACCOUNTING_OPS = (
@@ -161,12 +167,16 @@ def get_global_limits(self) -> ExecutionResources:
         default_limits = self._options.resource_limits
         exclude = self._options.exclude_resources
         cluster = ray.cluster_resources()
+        default_mem_fraction = (
+            self.DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION
+            if self.op_resource_allocator_enabled()
+            else self.DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION_WO_RESOURCE_RESERVATION
+        )
         cluster_resources = ExecutionResources(
             cpu=cluster.get("CPU", 0.0),
             gpu=cluster.get("GPU", 0.0),
             object_store_memory=round(
-                self.DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION
-                * cluster.get("object_store_memory", 0.0)
+                default_mem_fraction * cluster.get("object_store_memory", 0.0)
             ),
         )
         self._global_limits = default_limits.min(cluster_resources).subtract(exclude)