Repository: modin-project/modin (python)

Issue:
FEAT: Add 'auto' mode to the lazy execution.
The `MODIN_LAZY_EXECUTION` env var should have 3 options:
`Auto` - the execution mode is chosen by the engine for each operation (default value).
`On`   - the lazy execution is performed wherever it's possible.
`Off`  - the lazy execution is disabled.

Pull request description:
FEAT-#6918: Add auto mode to the lazy execution.
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #6918 <!-- issue must be created for each patch -->
- [x] tests added and passing
- [ ] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Patch:
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index 85bcf61e..56d821ba 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -816,18 +816,26 @@ class ReadSqlEngine(EnvironmentVariable, type=str):
     choices = ("Pandas", "Connectorx")
 
 
-class LazyExecution(EnvironmentVariable, type=bool):
-    """Prefer the lazy execution, when it's possible."""
+class LazyExecution(EnvironmentVariable, type=str):
+    """
+    Lazy execution mode.
+
+    Supported values:
+        `Auto` - the execution mode is chosen by the engine for each operation (default value).
+        `On`   - the lazy execution is performed wherever it's possible.
+        `Off`  - the lazy execution is disabled.
+    """
 
     varname = "MODIN_LAZY_EXECUTION"
-    default = False
+    choices = ("Auto", "On", "Off")
+    default = "Auto"
 
 
 class DaskThreadsPerWorker(EnvironmentVariable, type=int):
     """Number of threads per Dask worker."""
 
     varname = "MODIN_DASK_THREADS_PER_WORKER"
-    default = None
+    default = 1
 
 
 def _check_vars() -> None:
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
index 7c6bb38b..fc3f0b96 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py
@@ -125,11 +125,9 @@ class PandasOnRayDataframePartition(PandasDataframePartition):
         If ``LazyExecution`` is enabled, the function is not applied immediately,
         but is added to the execution tree.
         """
-        de = DeferredExecution(self._data_ref, func, args, kwargs)
-        if LazyExecution.get():
-            return self.__constructor__(de)
         log = get_logger()
         self._is_debug(log) and log.debug(f"ENTER::Partition.apply::{self._identity}")
+        de = DeferredExecution(self._data_ref, func, args, kwargs)
         data, meta, meta_offset = de.exec()
         self._is_debug(log) and log.debug(f"EXIT::Partition.apply::{self._identity}")
         return self.__constructor__(data, meta=meta, meta_offset=meta_offset)
@@ -377,3 +375,45 @@ def _get_index_and_columns(df):  # pragma: no cover
         The number of columns.
     """
     return len(df.index), len(df.columns)
+
+
+PandasOnRayDataframePartition._eager_exec_func = PandasOnRayDataframePartition.apply
+PandasOnRayDataframePartition._lazy_exec_func = (
+    PandasOnRayDataframePartition.add_to_apply_calls
+)
+
+
+def _configure_lazy_exec(cls: LazyExecution):
+    """Configure lazy execution mode for PandasOnRayDataframePartition."""
+    mode = cls.get()
+    get_logger().debug(f"Ray lazy execution mode: {mode}")
+    if mode == "Auto":
+        PandasOnRayDataframePartition.apply = (
+            PandasOnRayDataframePartition._eager_exec_func
+        )
+        PandasOnRayDataframePartition.add_to_apply_calls = (
+            PandasOnRayDataframePartition._lazy_exec_func
+        )
+    elif mode == "On":
+
+        def lazy_exec(self, func, *args, **kwargs):
+            return self._lazy_exec_func(func, *args, length=None, width=None, **kwargs)
+
+        PandasOnRayDataframePartition.apply = lazy_exec
+        PandasOnRayDataframePartition.add_to_apply_calls = (
+            PandasOnRayDataframePartition._lazy_exec_func
+        )
+    elif mode == "Off":
+
+        def eager_exec(self, func, *args, length=None, width=None, **kwargs):
+            return self._eager_exec_func(func, *args, **kwargs)
+
+        PandasOnRayDataframePartition.apply = (
+            PandasOnRayDataframePartition._eager_exec_func
+        )
+        PandasOnRayDataframePartition.add_to_apply_calls = eager_exec
+    else:
+        raise ValueError(f"Invalid lazy execution mode: {mode}")
+
+
+LazyExecution.subscribe(_configure_lazy_exec)


Test patch:
diff --git a/modin/test/storage_formats/pandas/test_internals.py b/modin/test/storage_formats/pandas/test_internals.py
index 1b7bf242..fe96a7c8 100644
--- a/modin/test/storage_formats/pandas/test_internals.py
+++ b/modin/test/storage_formats/pandas/test_internals.py
@@ -2421,3 +2421,50 @@ class TestZeroComputationDtypes:
             assert res_dtypes._known_dtypes["a"] == np.dtype("int64")
 
         patch.assert_not_called()
+
+
+@pytest.mark.skipif(Engine.get() != "Ray", reason="Ray specific")
+@pytest.mark.parametrize("mode", [None, "Auto", "On", "Off"])
+def test_ray_lazy_exec_mode(mode):
+    import ray
+
+    from modin.config import LazyExecution
+    from modin.core.execution.ray.common.deferred_execution import DeferredExecution
+    from modin.core.execution.ray.common.utils import ObjectIDType
+    from modin.core.execution.ray.implementations.pandas_on_ray.partitioning import (
+        PandasOnRayDataframePartition,
+    )
+
+    orig_mode = LazyExecution.get()
+    try:
+        if mode is None:
+            mode = LazyExecution.get()
+        else:
+            LazyExecution.put(mode)
+            assert mode == LazyExecution.get()
+
+        df = pandas.DataFrame({"A": [1, 2, 3]})
+        part = PandasOnRayDataframePartition(ray.put(df))
+
+        def func(df):
+            return len(df)
+
+        ray_func = ray.put(func)
+
+        if mode == "Auto":
+            assert isinstance(part.apply(ray_func)._data_ref, ObjectIDType)
+            assert isinstance(
+                part.add_to_apply_calls(ray_func)._data_ref, DeferredExecution
+            )
+        elif mode == "On":
+            assert isinstance(part.apply(ray_func)._data_ref, DeferredExecution)
+            assert isinstance(
+                part.add_to_apply_calls(ray_func)._data_ref, DeferredExecution
+            )
+        elif mode == "Off":
+            assert isinstance(part.apply(ray_func)._data_ref, ObjectIDType)
+            assert isinstance(part.add_to_apply_calls(ray_func)._data_ref, ObjectIDType)
+        else:
+            pytest.fail(f"Invalid value: {mode}")
+    finally:
+        LazyExecution.put(orig_mode)