FIX-#4564: Workaround import issues in Ray: auto-import pandas on python start if env var is set

docs/release_notes/release_notes-0.16.0.rst

@@ -13,6 +13,7 @@ Key Features and Updates
   * FIX-#4577: Set attribute of Modin dataframe to updated value (#4588)
   * FIX-#4411: Fix binary_op between datetime64 Series and pandas timedelta (#4592)
   * FIX-#4582: Inherit custom log layer (#4583)
+  * FIX-#4564: Workaround import issues in Ray: auto-import pandas on python start if env var is set (#4603)
 * Performance enhancements
   * PERF-#4182: Add cell-wise execution for binary ops, fix bin ops for empty dataframes (#4391)
 * Benchmarking enhancements

docs/requirements-doc.txt

@@ -9,12 +9,7 @@ pyyaml
 recommonmark
 sphinx
 sphinx-click
-# Pin ray to < 1.13.0 to work around GH#4564
-# TODO(https://github.com/modin-project/modin/issues/4564): let ray go past 1.13.0.
-ray[default]>=1.4.0,<1.13.0
-# Following https://github.com/ray-project/ray/pull/25648, pin protobuf < 4,
-# because versions >= 4.0.0 are incompatible with ray<1.13.0.
-protobuf<4.0.0
+ray[default]>=1.4.0
 git+https://github.com/modin-project/modin.git@master#egg=modin[all]
 sphinxcontrib_plantuml
 sphinx-issues

environment-dev.yml

@@ -41,12 +41,7 @@ dependencies:
       - modin-spreadsheet>=0.1.1
       - tqdm
       - git+https://github.com/airspeed-velocity/asv.git@ef016e233cb9a0b19d517135104f49e0a3c380e9
-      # Pin ray to < 1.13.0 to work around GH#4564
-      # TODO(https://github.com/modin-project/modin/issues/4564): let ray go past 1.13.0.
-      - ray[default]>=1.4.0,<1.13.0
-      # Following https://github.com/ray-project/ray/pull/25648, pin protobuf < 4,
-      # because versions >= 4.0.0 are incompatible with ray<1.13.0.
-      - protobuf<4.0.0
+      - ray[default]>=1.4.0
       - connectorx>=0.2.6a4
       # TODO: remove when resolving GH#4398
       - redis>=3.5.0,<4.0.0

modin-autoimport-pandas.pth

@@ -0,0 +1 @@
+import os; os.environ.get("__MODIN_AUTOIMPORT_PANDAS__", None) and __import__("pandas")

modin/config/envvars.py

@@ -83,7 +83,6 @@ def _get_default(cls):
         """
         from modin.utils import (
             MIN_RAY_VERSION,
-            MAX_RAY_VERSION_EXCLUSIVE,
             MIN_DASK_VERSION,
         )
 
@@ -95,14 +94,11 @@ def _get_default(cls):
         except ImportError:
             pass
         else:
-            if (
-                version.parse(ray.__version__) < MIN_RAY_VERSION
-                or version.parse(ray.__version__) >= MAX_RAY_VERSION_EXCLUSIVE
-            ):
+            if version.parse(ray.__version__) < MIN_RAY_VERSION:
                 raise ImportError(
                     "Please `pip install modin[ray]` to install compatible Ray "
                     + "version "
-                    + f"(>={MIN_RAY_VERSION},<{MAX_RAY_VERSION_EXCLUSIVE})."
+                    + f"(>={MIN_RAY_VERSION})."
                 )
             return "Ray"
         try:

modin/core/execution/ray/common/utils.py

@@ -32,6 +32,7 @@
     NPartitions,
     ValueSource,
 )
+from modin.error_message import ErrorMessage
 
 ObjectIDType = ray.ObjectRef
 if version.parse(ray.__version__) >= version.parse("1.2.0"):
@@ -40,60 +41,6 @@
     ObjectIDType = (ray.ObjectRef, ClientObjectRef)
 
 
-def _move_stdlib_ahead_of_site_packages(*args):
-    """
-    Ensure packages from stdlib have higher import priority than from site-packages.
-
-    Parameters
-    ----------
-    *args : tuple
-        Ignored, added for compatibility with Ray.
-
-    Notes
-    -----
-    This function is expected to be run on all workers including the driver.
-    This is a hack solution to fix GH-#647, GH-#746.
-    """
-    site_packages_path = None
-    site_packages_path_index = -1
-    for i, path in enumerate(sys.path):
-        if sys.exec_prefix in path and path.endswith("site-packages"):
-            site_packages_path = path
-            site_packages_path_index = i
-            # break on first found
-            break
-
-    if site_packages_path is not None:
-        # stdlib packages layout as follows:
-        # - python3.x
-        #   - typing.py
-        #   - site-packages/
-        #     - pandas
-        # So extracting the dirname of the site_packages can point us
-        # to the directory containing standard libraries.
-        sys.path.insert(site_packages_path_index, os.path.dirname(site_packages_path))
-
-
-def _import_pandas(*args):
-    """
-    Import pandas to make sure all its machinery is ready.
-
-    This prevents a race condition between two threads deserializing functions
-    and trying to import pandas at the same time.
-
-    Parameters
-    ----------
-    *args : tuple
-        Ignored, added for compatibility with Ray.
-
-    Notes
-    -----
-    This function is expected to be run on all workers before any
-    serialization or deserialization starts.
-    """
-    import pandas  # noqa F401
-
-
 def initialize_ray(
     override_is_cluster=False,
     override_redis_address: str = None,
@@ -117,6 +64,7 @@ def initialize_ray(
         What password to use when connecting to Redis.
         If not specified, ``modin.config.RayRedisPassword`` is used.
     """
+    extra_init_kw = {"runtime_env": {"env_vars": {"__MODIN_AUTOIMPORT_PANDAS__": "1"}}}
     if not ray.is_initialized() or override_is_cluster:
         cluster = override_is_cluster or IsRayCluster.get()
         redis_address = override_redis_address or RayRedisAddress.get()
@@ -138,17 +86,16 @@ def initialize_ray(
                 include_dashboard=False,
                 ignore_reinit_error=True,
                 _redis_password=redis_password,
+                **extra_init_kw,
             )
         else:
-            from modin.error_message import ErrorMessage
-
             # This string is intentionally formatted this way. We want it indented in
             # the warning message.
             ErrorMessage.not_initialized(
                 "Ray",
-                """
+                f"""
     import ray
-    ray.init()
+    ray.init({', '.join([f'{k}={v}' for k,v in extra_init_kw.items()])})
 """,
             )
             object_store_memory = Memory.get()
@@ -206,6 +153,7 @@ def initialize_ray(
                 "object_store_memory": object_store_memory,
                 "_redis_password": redis_password,
                 "_memory": object_store_memory,
+                **extra_init_kw,
             }
             ray.init(**ray_init_kwargs)
 
@@ -219,11 +167,15 @@ def initialize_ray(
             if not GPU_MANAGERS:
                 for i in range(GpuCount.get()):
                     GPU_MANAGERS.append(GPUManager.remote(i))
-    _move_stdlib_ahead_of_site_packages()
-    ray.worker.global_worker.run_function_on_all_workers(
-        _move_stdlib_ahead_of_site_packages
-    )
-    ray.worker.global_worker.run_function_on_all_workers(_import_pandas)
+    else:  # ray is already initialized, check runtime env config
+        env_vars = ray.get_runtime_context().runtime_env.get("env_vars", {})
+        for varname, varvalue in extra_init_kw["runtime_env"]["env_vars"].items():
+            if str(env_vars.get(varname, "")) != str(varvalue):
+                ErrorMessage.single_warning(
+                    "If initialising Ray yourself, please ensure its runtime env "
+                    + f"sets environment variable {varname} to {varvalue}"
+                )
+
     num_cpus = int(ray.cluster_resources()["CPU"])
     num_gpus = int(ray.cluster_resources().get("GPU", 0))
     if StorageFormat.get() == "Cudf":

modin/utils.py

@@ -36,9 +36,6 @@
 from modin._version import get_versions
 
 MIN_RAY_VERSION = version.parse("1.4.0")
-# TODO(https://github.com/modin-project/modin/issues/4564):
-# once ray can go past 1.13.0, remove this constant and stop checking it.
-MAX_RAY_VERSION_EXCLUSIVE = version.parse("1.13.0")
 MIN_DASK_VERSION = version.parse("2.22.0")
 
 PANDAS_API_URL_TEMPLATE = f"https://pandas.pydata.org/pandas-docs/version/{pandas.__version__}/reference/api/{{}}.html"

requirements-dev.txt

@@ -3,12 +3,7 @@ numpy>=1.18.5
 pyarrow>=4.0.1
 dask[complete]>=2.22.0,<2022.2.0
 distributed>=2.22.0,<2022.2.0
-# Pin ray to < 1.13.0 to work around GH#4564
-# TODO(https://github.com/modin-project/modin/issues/4564): let ray go past 1.13.0.
-ray[default]>=1.4.0,<1.13.0
-# Following https://github.com/ray-project/ray/pull/25648, pin protobuf < 4,
-# because versions >= 4.0.0 are incompatible with ray<1.13.0.
-protobuf<4.0.0
+ray[default]>=1.4.0
 redis>=3.5.0,<4.0.0
 psutil
 fsspec

setup.py

@@ -5,26 +5,42 @@
     long_description = fh.read()
 
 dask_deps = ["dask>=2.22.0,<2022.2.0", "distributed>=2.22.0,<2022.2.0"]
-# TODO: remove redis dependency when resolving GH#4398
-# Pin ray to < 1.13.0 to work around GH#4564
-# TODO(https://github.com/modin-project/modin/issues/4564): let ray go past 1.13.0.
-# Following https://github.com/ray-project/ray/pull/25648, pin protobuf < 4,
-# because versions >= 4.0.0 are incompatible with ray<1.13.0.
 ray_deps = [
-    "ray[default]>=1.4.0,<1.13.0",
+    "ray[default]>=1.4.0",
     "pyarrow>=4.0.1",
     "redis>=3.5.0,<4.0.0",
-    "protobuf<4.0.0",
 ]
 remote_deps = ["rpyc==4.1.5", "cloudpickle", "boto3"]
 spreadsheet_deps = ["modin-spreadsheet>=0.1.0"]
 sql_deps = ["dfsql>=0.4.2", "pyparsing<=2.4.7"]
 all_deps = dask_deps + ray_deps + remote_deps + spreadsheet_deps
 
+# Distribute 'modin-autoimport-pandas.pth' along with binary and source distributions.
+# This file provides the "import pandas before Ray init" feature if specific
+# environment variable is set (see https://github.com/modin-project/modin/issues/4564).
+cmdclass = versioneer.get_cmdclass()
+
+
+class AddPthFileBuild(cmdclass["build_py"]):
+    def _get_data_files(self):
+        return (super()._get_data_files() or []) + [
+            (".", ".", self.build_lib, ["modin-autoimport-pandas.pth"])
+        ]
+
+
+class AddPthFileSDist(cmdclass["sdist"]):
+    def make_distribution(self):
+        self.filelist.append("modin-autoimport-pandas.pth")
+        return super().make_distribution()
+
+
+cmdclass["build_py"] = AddPthFileBuild
+cmdclass["sdist"] = AddPthFileSDist
+
 setup(
     name="modin",
     version=versioneer.get_version(),
-    cmdclass=versioneer.get_cmdclass(),
+    cmdclass=cmdclass,
     description="Modin: Make your pandas code run faster by changing one line of code.",
     packages=find_packages(exclude=["scripts", "scripts.*"]),
     include_package_data=True,