[Data] Set default strategy for callable class in get_compute_strategy (ray-project#57657)

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

## Why are these changes needed?
Instead of raising ValueError when `compute` and `concurrency` is unset
for callable class, we use `ActorPoolStrategy(min_size=1,
max_size=None)` as default value.
<!-- Please give a short summary of the change and the problem this
solves. -->

## Related issue number
Follow up for ray-project#57035
<!-- For example: "Closes ray-project#1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run pre-commit jobs to lint the changes in this PR.
([pre-commit
setup](https://docs.ray.io/en/latest/ray-contribute/getting-involved.html#lint-and-formatting))
- [ ] 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
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: You-Cheng Lin (Owen) <mses010108@gmail.com>

Author: owenowenisme

python/ray/data/_internal/util.py

@@ -577,17 +577,19 @@ def get_compute_strategy(
             compute == "tasks" or isinstance(compute, TaskPoolStrategy)
         ):
             raise ValueError(
-                "``compute`` must specify an actor compute strategy when using a "
-                f"callable class, but got: {compute}. For example, use "
-                "``compute=ray.data.ActorPoolStrategy(size=n)``."
+                f"You specified the callable class {fn} as your UDF with the compute "
+                f"{compute}, but Ray Data can't schedule callable classes with the task "
+                f"pool strategy. To fix this error, pass an ActorPoolStrategy to compute or "
+                f"None to use the default compute strategy."
             )
         elif not is_callable_class and (
             compute == "actors" or isinstance(compute, ActorPoolStrategy)
         ):
             raise ValueError(
-                f"``compute`` is specified as the actor compute strategy: {compute}, "
-                f"but ``fn`` is not a callable class: {fn}. Pass a callable class or "
-                "use the default ``compute`` strategy."
+                f"You specified the function {fn} as your UDF with the compute "
+                f"{compute}, but Ray Data can't schedule regular functions with the actor "
+                f"pool strategy. To fix this error, pass a TaskPoolStrategy to compute or "
+                f"None to use the default compute strategy."
             )
         return compute
     elif concurrency is not None:
@@ -639,10 +641,7 @@ def get_compute_strategy(
             )
     else:
         if is_callable_class:
-            raise ValueError(
-                "``concurrency`` must be specified when using a callable class. "
-                "For example, use ``concurrency=n`` for a pool of ``n`` workers."
-            )
+            return ActorPoolStrategy(min_size=1, max_size=None)
         else:
             return TaskPoolStrategy()
 

python/ray/data/dataset.py

@@ -343,10 +343,12 @@ def parse_filename(row: Dict[str, Any]) -> Dict[str, Any]:
                 that can be instantiated to create such a callable.
             compute: The compute strategy to use for the map operation.
 
-                * If ``compute`` is not specified, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
+                * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
 
                 * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks.
 
+                * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers.
+
                 * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers.
 
                 * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers.
@@ -583,10 +585,12 @@ def __call__(self, batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
                 to a given map task. Default ``batch_size`` is ``None``.
             compute: The compute strategy to use for the map operation.
 
-                * If ``compute`` is not specified, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
+                * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
 
                 * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks.
 
+                * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers.
+
                 * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers.
 
                 * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers.
@@ -1288,10 +1292,12 @@ def duplicate_row(row: Dict[str, Any]) -> List[Dict[str, Any]]:
                 that can be instantiated to create such a callable.
             compute: The compute strategy to use for the map operation.
 
-                * If ``compute`` is not specified, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
+                * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
 
                 * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks.
 
+                * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers.
+
                 * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers.
 
                 * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers.
@@ -1426,10 +1432,12 @@ def filter(
                 are top-level arguments in the underlying Ray actor construction task.
             compute: The compute strategy to use for the map operation.
 
-                * If ``compute`` is not specified, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
+                * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
 
                 * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks.
 
+                * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers.
+
                 * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers.
 
                 * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers.

python/ray/data/grouped_data.py

@@ -163,10 +163,12 @@ def map_groups(
                 making an additional copy.
             compute: The compute strategy to use for the map operation.
 
-                * If ``compute`` is not specified, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
+                * If ``compute`` is not specified for a function, will use ``ray.data.TaskPoolStrategy()`` to launch concurrent tasks based on the available resources and number of input blocks.
 
                 * Use ``ray.data.TaskPoolStrategy(size=n)`` to launch at most ``n`` concurrent Ray tasks.
 
+                * If ``compute`` is not specified for a callable class, will use ``ray.data.ActorPoolStrategy(min_size=1, max_size=None)`` to launch an autoscaling actor pool from 1 to unlimited workers.
+
                 * Use ``ray.data.ActorPoolStrategy(size=n)`` to use a fixed size actor pool of ``n`` workers.
 
                 * Use ``ray.data.ActorPoolStrategy(min_size=m, max_size=n)`` to use an autoscaling actor pool from ``m`` to ``n`` workers.

python/ray/data/tests/test_map.py

@@ -307,13 +307,6 @@ def __call__(self, x):
         with pytest.raises(ValueError, match=error_message):
             ds.map(UDFClass, concurrency=concurrency).take_all()
 
-    # Test concurrency not set.
-    result = ds.map(udf).take_all()
-    assert sorted(extract_values("id", result)) == list(range(10)), result
-    error_message = "``concurrency`` must be specified when using a callable class."
-    with pytest.raises(ValueError, match=error_message):
-        ds.map(UDFClass).take_all()
-
 
 @pytest.mark.parametrize("udf_kind", ["gen", "func"])
 def test_flat_map(