fix(core): write use_polars_sort to Ray DatasetContext

docs/changelog.md

@@ -6,6 +6,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## Unreleased
+
+### :bug: Bug Fixes
+
+- avoid Ray 2.55 `DatasetContext.use_polars` DeprecationWarning on every `RayDataExecutionOptions.apply()` by writing to `use_polars_sort` ([#354](https://github.com/danielgafni/dagster-ray/issues/354))
+
+### :hammer_and_wrench: Other Improvements
+
+- add `RayDataExecutionOptions.use_polars_sort`; keep `use_polars` as a deprecated alias that forwards its value and emits a `DeprecationWarning`
+
 ## v0.4.4 (26-03-2026)
 
 ### :sparkles: Features

src/dagster_ray/configs.py

@@ -4,7 +4,8 @@
 from typing import Any, Literal
 
 import dagster as dg
-from pydantic import Field
+from packaging.version import Version
+from pydantic import Field, model_validator
 
 USER_DEFINED_RAY_KEY = "dagster-ray/config"
 DAGSTER_RAY_NAMESPACES_ENV_VAR = "DAGSTER_RAY_NAMESPACES"
@@ -77,7 +78,32 @@ class RayDataExecutionOptions(dg.Config):
     cpu_limit: int = 5000
     gpu_limit: int = 0
     verbose_progress: bool = True
-    use_polars: bool = True
+    use_polars_sort: bool = Field(
+        default=True,
+        description="Whether to use Polars for sort operations in Ray Data. Forwarded to `ray.data.DatasetContext.use_polars_sort`.",
+    )
+    use_polars: bool | None = Field(
+        default=None,
+        description="Deprecated. Use `use_polars_sort` instead. Ray 2.55 renamed the underlying `DatasetContext` attribute to `use_polars_sort`.",
+        deprecated="`use_polars` is deprecated; use `use_polars_sort` instead.",
+    )
+
+    @model_validator(mode="before")
+    @classmethod
+    def _forward_deprecated_use_polars(cls, data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+        legacy = data.get("use_polars")
+        new = data.get("use_polars_sort")
+        if legacy is not None and new is not None and legacy != new:
+            raise ValueError(
+                "`use_polars` and `use_polars_sort` were set to contradicting values "
+                f"({legacy!r} vs {new!r}). `use_polars` is deprecated; set only "
+                "`use_polars_sort`."
+            )
+        if legacy is not None and new is None:
+            data = {**data, "use_polars_sort": legacy}
+        return data
 
     def apply(self):
         import ray
@@ -92,7 +118,14 @@ def apply(self):
         )
 
         ctx.verbose_progress = self.verbose_progress
-        ctx.use_polars = self.use_polars
+        # Ray >=2.50 renamed the sort-time Polars toggle from `use_polars` to
+        # `use_polars_sort`; read the effective value (the legacy field wins when
+        # both are set only because the validator has already unified them).
+        effective = self.use_polars if self.use_polars is not None else self.use_polars_sort
+        if Version(ray.__version__) >= Version("2.50"):
+            ctx.use_polars_sort = effective
+        else:
+            ctx.use_polars = effective
 
     def apply_remote(self):
         import ray

tests/test_configs.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import warnings
+
+import pytest
+from pydantic import ValidationError
+
+from dagster_ray.configs import RayDataExecutionOptions
+
+
+def test_use_polars_sort_defaults_to_true():
+    opts = RayDataExecutionOptions()
+    assert opts.use_polars_sort is True
+    # accessing the deprecated field emits a warning, so silence it for this assertion
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        assert opts.use_polars is None
+
+
+def test_deprecated_use_polars_is_forwarded_to_use_polars_sort():
+    opts = RayDataExecutionOptions(use_polars=False)
+    assert opts.use_polars_sort is False
+    # reading the deprecated attribute warns and preserves backward compatibility
+    with pytest.warns(DeprecationWarning, match="use_polars_sort"):
+        assert opts.use_polars is False
+
+
+def test_use_polars_and_use_polars_sort_in_agreement_is_accepted():
+    opts = RayDataExecutionOptions(use_polars=True, use_polars_sort=True)
+    assert opts.use_polars_sort is True
+
+
+def test_use_polars_and_use_polars_sort_contradicting_raises():
+    with pytest.raises(ValidationError, match="contradicting values"):
+        RayDataExecutionOptions(use_polars=True, use_polars_sort=False)
+    with pytest.raises(ValidationError, match="contradicting values"):
+        RayDataExecutionOptions(use_polars=False, use_polars_sort=True)
+
+
+def test_use_polars_sort_alone_does_not_warn_on_construction():
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        opts = RayDataExecutionOptions(use_polars_sort=False)
+    ours = [w for w in caught if "use_polars" in str(w.message)]
+    assert ours == []
+    assert opts.use_polars_sort is False