[core] Unify RuntimeEnv archive validation and support tar.xz

doc/source/ray-core/handling-dependencies.rst

@@ -493,7 +493,7 @@ API Reference
 
 The ``runtime_env`` is a Python dictionary or a Python class :class:`ray.runtime_env.RuntimeEnv <ray.runtime_env.RuntimeEnv>` including one or more of the following fields:
 
-- ``working_dir`` (str): Specifies the working directory for the Ray workers. This must either be (1) a local existing directory with total size at most 500 MiB, (2) a local existing archive file (``.zip``, ``.tar.gz``, or ``.tgz``) with total uncompressed size at most 500 MiB (Note: ``excludes`` has no effect), or (3) a URI to a remotely-stored archive (``.zip``, ``.tar.gz``, or ``.tgz``) containing the working directory for your job (no file size limit is enforced by Ray). See :ref:`remote-uris` for details.
+- ``working_dir`` (str): Specifies the working directory for the Ray workers. This must either be (1) a local existing directory with total size at most 500 MiB, (2) a local existing archive file (``.zip``, ``.tar.gz``, ``.tgz``, or ``.tar.xz``) with total uncompressed size at most 500 MiB (Note: ``excludes`` has no effect), or (3) a URI to a remotely-stored archive (``.zip``, ``.tar.gz``, ``.tgz``, or ``.tar.xz``) containing the working directory for your job (no file size limit is enforced by Ray). See :ref:`remote-uris` for details.
   The specified directory is downloaded to each node on the cluster, and Ray workers start in their node's copy of this directory.
 
   - Examples
@@ -517,7 +517,7 @@ The ``runtime_env`` is a Python dictionary or a Python class :class:`ray.runtime
   Note: If the local directory contains symbolic links, Ray follows the links and the files they point to are uploaded to the cluster.
 
 - ``py_modules`` (List[str|module]): Specifies Python modules to be available for import in the Ray workers.  (For more ways to specify packages, see also the ``pip`` and ``conda`` fields below.)
-  Each entry must be either (1) a path to a local file or directory, (2) a URI to a remote archive (``.zip``, ``.tar.gz``, ``.tgz``) or wheel (``.whl``) file (see :ref:`remote-uris` for details), (3) a Python module object, or (4) a path to a local ``.whl`` file.
+  Each entry must be either (1) a path to a local file or directory, (2) a URI to a remote archive (``.zip``, ``.tar.gz``, ``.tgz``, ``.tar.xz``) or wheel (``.whl``) file (see :ref:`remote-uris` for details), (3) a Python module object, or (4) a path to a local ``.whl`` file.
 
   - Examples of entries in the list:
 
@@ -828,7 +828,7 @@ If you want to specify this directory as a local path, your ``runtime_env`` dict
   runtime_env = {..., "working_dir": "/some_path/example_dir", ...}
 
 Suppose instead you want to host your files in your ``/some_path/example_dir`` directory remotely and provide a remote URI.
-You need to first compress the ``example_dir`` directory into a ``.zip`` or ``.tar.gz`` archive.
+You need to first compress the ``example_dir`` directory into a ``.zip``, ``.tar.gz``, ``.tgz``, or ``.tar.xz`` archive.
 
 There should be no other files or directories at the top level of the archive, other than ``example_dir``.
 You can use one of the following commands in the Terminal:
@@ -840,6 +840,8 @@ You can use one of the following commands in the Terminal:
     zip -r archive.zip example_dir
     # Using tar.gz:
     tar -czf archive.tar.gz example_dir
+    # Using tar.xz:
+    tar -cJf archive.tar.xz example_dir
 
 Run this command from the *parent directory* of the desired ``working_dir`` to ensure that the resulting archive contains a single top-level directory.
 In general, the archive's name and the top-level directory's name can be anything.
@@ -853,6 +855,8 @@ You can check that the archive contains a single top-level directory by running
   zipinfo -1 archive.zip
   # For tar.gz:
   tar -tzf archive.tar.gz
+  # For tar.xz:
+  tar -tJf archive.tar.xz
   # example_dir/
   # example_dir/my_file_1.txt
   # example_dir/subdir/my_file_2.txt
@@ -865,13 +869,20 @@ Your ``runtime_env`` dictionary should contain:
 
   runtime_env = {..., "working_dir": "s3://example_bucket/example.zip", ...}
 
-You can also use ``.tar.gz`` or ``.tgz`` archives:
+You can also use ``.tar.gz``, ``.tgz``, or ``.tar.xz`` archives:
 
 .. testcode::
   :skipif: True
 
   runtime_env = {..., "working_dir": "s3://example_bucket/example.tar.gz", ...}
 
+For example, an XZ-compressed archive can be specified in the same way:
+
+.. testcode::
+  :skipif: True
+
+  runtime_env = {..., "working_dir": "s3://example_bucket/example.tar.xz", ...}
+
 .. warning::
 
   Check for hidden files and metadata directories in archived dependencies.
@@ -880,7 +891,7 @@ You can also use ``.tar.gz`` or ``.tgz`` archives:
   To avoid this, use ``zip -r`` or ``tar -czf`` directly on the directory you want to compress from its parent's directory. For example, if you have a directory structure such as: ``a/b`` and you want to compress ``b``, issue the command from the directory ``a``.
   If Ray detects more than a single directory at the top level, it uses the entire archive instead of the top-level directory, which may lead to unexpected behavior.
 
-Remote URIs support ``.zip``, ``.tar.gz``, and ``.tgz`` archive formats. Four types of remote URIs are supported for hosting ``working_dir`` and ``py_modules`` packages:
+Remote URIs support ``.zip``, ``.tar.gz``, ``.tgz``, and ``.tar.xz`` archive formats. Supported schemes are ``http``, ``https``, ``s3``, ``gs``, ``azure``, ``abfss``, and ``file``. The most common remote storage types are described below:
 
 - ``HTTPS``: ``HTTPS`` refers to URLs that start with ``https``.
   These are particularly useful because remote Git providers (e.g. GitHub, Bitbucket, GitLab, etc.) use ``https`` URLs as download links for repository archives.

python/ray/_common/runtime_env_package.py

@@ -0,0 +1,102 @@
+from typing import Dict, Optional, Tuple
+
+WORKING_DIR = "working_dir"
+PY_MODULES = "py_modules"
+
+ZIP_EXTENSION = ".zip"
+WHEEL_EXTENSION = ".whl"
+JAR_EXTENSION = ".jar"
+TAR_GZ_EXTENSION = ".tar.gz"
+TGZ_EXTENSION = ".tgz"
+TAR_BZ2_EXTENSION = ".tar.bz2"
+TAR_XZ_EXTENSION = ".tar.xz"
+
+# These are the user-facing formats accepted for RuntimeEnv package fields.
+# Keep field-specific validation, local uploads, and Job package uploads derived
+# from this mapping so they cannot silently diverge.
+RUNTIME_ENV_PACKAGE_EXTENSIONS: Dict[str, Tuple[str, ...]] = {
+    WORKING_DIR: (
+        ZIP_EXTENSION,
+        TAR_GZ_EXTENSION,
+        TGZ_EXTENSION,
+        TAR_XZ_EXTENSION,
+    ),
+    PY_MODULES: (
+        ZIP_EXTENSION,
+        WHEEL_EXTENSION,
+        TAR_GZ_EXTENSION,
+        TGZ_EXTENSION,
+        TAR_XZ_EXTENSION,
+    ),
+}
+
+PACKAGE_UPLOAD_EXTENSIONS = tuple(
+    dict.fromkeys(
+        extension
+        for extensions in RUNTIME_ENV_PACKAGE_EXTENSIONS.values()
+        for extension in extensions
+    )
+)
+
+# .tar.bz2 is retained as a low-level download format for compatibility, but it
+# is not part of the public working_dir or py_modules contract above.
+TAR_EXTENSIONS = (
+    TAR_GZ_EXTENSION,
+    TGZ_EXTENSION,
+    TAR_BZ2_EXTENSION,
+    TAR_XZ_EXTENSION,
+)
+COMPOUND_ARCHIVE_EXTENSIONS = (
+    TAR_GZ_EXTENSION,
+    TAR_BZ2_EXTENSION,
+    TAR_XZ_EXTENSION,
+)
+
+
+def get_package_extension(
+    path: str, supported_extensions: Tuple[str, ...]
+) -> Optional[str]:
+    """Return the supported extension for a package path.
+
+    Args:
+        path: Package path or URI to inspect.
+        supported_extensions: Extensions to match in any order, including compound
+            extensions. If extensions overlap, the longest match takes precedence.
+
+    Returns:
+        The longest matching extension, or ``None`` if the path is unsupported.
+    """
+    for extension in sorted(supported_extensions, key=len, reverse=True):
+        if path.endswith(extension):
+            return extension
+    return None
+
+
+def has_package_extension(path: str, supported_extensions: Tuple[str, ...]) -> bool:
+    """Return whether a package path ends in a supported extension."""
+    return get_package_extension(path, supported_extensions) is not None
+
+
+def validate_package_extension(
+    path: str, field: str, display_path: Optional[str] = None
+) -> None:
+    """Validate a RuntimeEnv package path using the field's capabilities.
+
+    Args:
+        path: Package path to validate.
+        field: RuntimeEnv field whose format capabilities apply.
+        display_path: Optional sanitized user-facing path to include in validation
+            errors. URI query parameters must be removed before passing it.
+
+    Raises:
+        ValueError: If the path does not have an extension supported by the field.
+    """
+    supported_extensions = RUNTIME_ENV_PACKAGE_EXTENSIONS[field]
+    if has_package_extension(path, supported_extensions):
+        return
+
+    formats = ", ".join(supported_extensions)
+    error_path = path if display_path is None else display_path
+    raise ValueError(
+        f"Only {formats} files are supported for {field} URIs; got {error_path}."
+    )

python/ray/_common/runtime_env_uri.py

@@ -5,6 +5,12 @@
 from typing import Tuple
 from urllib.parse import urlparse
 
+from ray._common.runtime_env_package import (
+    COMPOUND_ARCHIVE_EXTENSIONS,
+    WHEEL_EXTENSION,
+    get_package_extension,
+)
+
 _REMOTE_PROTOCOLS = ("http", "https", "s3", "gs", "azure", "abfss", "file")
 
 
@@ -35,8 +41,7 @@ class Protocol(enum.Enum):
     @classmethod
     def remote_protocols(cls):
         # Returns a list of protocols that support remote storage.
-        # These protocols should only be used with paths that end in
-        # ".zip", ".whl", ".tar.gz", or ".tgz".
+        # RuntimeEnv fields apply their own format constraints to these URIs.
         return [cls[protocol.upper()] for protocol in _REMOTE_PROTOCOLS]
 
 
@@ -83,7 +88,7 @@ def parse_uri(pkg_uri: str) -> Tuple[Protocol, str]:
         )
 
     if protocol in Protocol.remote_protocols():
-        if uri.path.endswith(".whl"):
+        if uri.path.endswith(WHEEL_EXTENSION):
             # Don't modify the .whl filename. See
             # https://peps.python.org/pep-0427/#file-name-convention
             # for more information.
@@ -92,16 +97,13 @@ def parse_uri(pkg_uri: str) -> Tuple[Protocol, str]:
             # Hash the URI to produce a stable, NAME_MAX-safe local filename
             # regardless of how long or deeply nested the URI is. The extension
             # is preserved so is_zip_uri / is_jar_uri keep working. Compound
-            # extensions (.tar.gz, .tar.bz2) are kept intact so archive-type
+            # compound extensions are kept intact so archive-type
             # detection downstream still works.
             # netloc + path covers URIs where the filename has no path
             # component (e.g., s3://package.zip puts "package.zip" in netloc).
             raw = uri.netloc + uri.path
-            if raw.endswith(".tar.gz"):
-                suffix = ".tar.gz"
-            elif raw.endswith(".tar.bz2"):
-                suffix = ".tar.bz2"
-            else:
+            suffix = get_package_extension(raw, COMPOUND_ARCHIVE_EXTENSIONS)
+            if suffix is None:
                 suffix = pathlib.Path(raw).suffix
             digest = hashlib.sha1(pkg_uri.encode("utf-8")).hexdigest()
             package_name = f"{protocol.value}_{digest}{suffix}"

python/ray/_common/tests/test_runtime_env_uri.py

@@ -201,7 +201,7 @@ def test_parse_uri_remote_paths_are_unique(self):
             "s3://package{ext}",  # extension in netloc (no path component)
         ],
     )
-    @pytest.mark.parametrize("ext", [".zip", ".tar.gz", ".tar.bz2"])
+    @pytest.mark.parametrize("ext", [".zip", ".tar.gz", ".tar.bz2", ".tar.xz"])
     def test_parse_uri_remote_preserves_extension(self, uri_template, ext):
         """Extensions are kept intact after hashing, whether the filename
         is in the path or the netloc."""

python/ray/_private/runtime_env/packaging.py

@@ -14,6 +14,18 @@
 
 from filelock import FileLock
 
+from ray._common.runtime_env_package import (
+    COMPOUND_ARCHIVE_EXTENSIONS,
+    JAR_EXTENSION,
+    TAR_EXTENSIONS,
+    TAR_GZ_EXTENSION,
+    TAR_XZ_EXTENSION,
+    TGZ_EXTENSION,
+    WHEEL_EXTENSION,
+    ZIP_EXTENSION,
+    get_package_extension,
+    has_package_extension,
+)
 from ray._common.runtime_env_uri import parse_uri as _parse_uri
 from ray._private.ray_constants import (
     GRPC_CPP_MAX_MESSAGE_SIZE,
@@ -259,7 +271,7 @@ def is_zip_uri(uri: str) -> bool:
     except ValueError:
         return False
 
-    return Path(path).suffix == ".zip"
+    return Path(path).suffix == ZIP_EXTENSION
 
 
 def is_whl_uri(uri: str) -> bool:
@@ -268,7 +280,7 @@ def is_whl_uri(uri: str) -> bool:
     except ValueError:
         return False
 
-    return Path(path).suffix == ".whl"
+    return Path(path).suffix == WHEEL_EXTENSION
 
 
 def is_jar_uri(uri: str) -> bool:
@@ -277,7 +289,7 @@ def is_jar_uri(uri: str) -> bool:
     except ValueError:
         return False
 
-    return Path(path).suffix == ".jar"
+    return Path(path).suffix == JAR_EXTENSION
 
 
 def is_tar_gz_uri(uri: str) -> bool:
@@ -286,7 +298,16 @@ def is_tar_gz_uri(uri: str) -> bool:
     except ValueError:
         return False
 
-    return path.endswith(".tar.gz") or Path(path).suffix == ".tgz"
+    return path.endswith(TAR_GZ_EXTENSION) or Path(path).suffix == TGZ_EXTENSION
+
+
+def is_tar_uri(uri: str) -> bool:
+    try:
+        _, path = _parse_uri(uri)
+    except ValueError:
+        return False
+
+    return has_package_extension(path, TAR_EXTENSIONS)
 
 
 def _get_excludes(path: Path, excludes: List[str]) -> Callable:
@@ -525,20 +546,22 @@ def package_exists(pkg_uri: str) -> bool:
 def get_uri_for_package(package: Path) -> str:
     """Get a content-addressable URI from a package's contents."""
 
-    if package.suffix == ".whl":
+    if package.suffix == WHEEL_EXTENSION:
         # Wheel file names include the Python package name, version
         # and tags, so it is already effectively content-addressed.
         return "{protocol}://{whl_filename}".format(
             protocol=Protocol.GCS.value, whl_filename=package.name
         )
     else:
         hash_val = hashlib.sha1(package.read_bytes()).hexdigest()
-        if package.name.endswith(".tar.gz"):
-            ext = ".tar.gz"
-        elif package.suffix == ".tgz":
-            ext = ".tar.gz"
+        if package.name.endswith(TAR_GZ_EXTENSION):
+            ext = TAR_GZ_EXTENSION
+        elif package.suffix == TGZ_EXTENSION:
+            ext = TAR_GZ_EXTENSION
+        elif package.name.endswith(TAR_XZ_EXTENSION):
+            ext = TAR_XZ_EXTENSION
         else:
-            ext = ".zip"
+            ext = ZIP_EXTENSION
         return "{protocol}://{pkg_name}{ext}".format(
             protocol=Protocol.GCS.value, pkg_name=RAY_PKG_PREFIX + hash_val, ext=ext
         )
@@ -818,10 +841,9 @@ def get_local_dir_from_uri(uri: str, base_directory: str) -> Path:
     """Return the local directory corresponding to this URI."""
     pkg_file = Path(_get_local_path(base_directory, uri))
     pkg_name = pkg_file.name
-    if pkg_name.endswith(".tar.gz"):
-        local_dir = pkg_file.parent / pkg_name[: -len(".tar.gz")]
-    elif pkg_name.endswith(".tar.bz2"):
-        local_dir = pkg_file.parent / pkg_name[: -len(".tar.bz2")]
+    compound_extension = get_package_extension(pkg_name, COMPOUND_ARCHIVE_EXTENSIONS)
+    if compound_extension is not None:
+        local_dir = pkg_file.parent / pkg_name[: -len(compound_extension)]
     else:
         local_dir = pkg_file.with_suffix("")
     return local_dir
@@ -837,7 +859,8 @@ async def download_and_unpack_package(
 ) -> str:
     """Download the package corresponding to this URI and unpack it.
 
-    Supports .zip, .jar, .tar.gz, and .tgz archives for remote protocols.
+    Supports .zip, .jar, .tar.gz, .tgz, .tar.bz2, and .tar.xz archives for
+    remote protocols.
     Will be written to a file or directory named {base_directory}/{uri}.
     Returns the path to this file or directory.
 
@@ -927,7 +950,7 @@ async def download_and_unpack_package(
                         unlink_zip=True,
                         logger=logger,
                     )
-                elif is_tar_gz_uri(pkg_uri):
+                elif is_tar_uri(pkg_uri):
                     untar_package(
                         package_path=pkg_file,
                         target_dir=local_dir,
@@ -940,21 +963,17 @@ async def download_and_unpack_package(
             elif protocol in Protocol.remote_protocols():
                 protocol.download_remote_uri(source_uri=pkg_uri, dest_file=pkg_file)
 
-                if pkg_file.suffix in [".zip", ".jar"]:
+                if pkg_file.suffix in [ZIP_EXTENSION, JAR_EXTENSION]:
                     unzip_package(
                         package_path=pkg_file,
                         target_dir=local_dir,
                         remove_top_level_directory=True,
                         unlink_zip=True,
                         logger=logger,
                     )
-                elif pkg_file.suffix == ".whl":
+                elif pkg_file.suffix == WHEEL_EXTENSION:
                     return str(pkg_file)
-                elif (
-                    str(pkg_file).endswith(".tar.gz")
-                    or pkg_file.suffix == ".tgz"
-                    or str(pkg_file).endswith(".tar.bz2")
-                ):
+                elif is_tar_uri(pkg_uri):
                     untar_package(
                         package_path=pkg_file,
                         target_dir=local_dir,