From 4c7c064ee0b3f23a5497df781743ba24aa615738 Mon Sep 17 00:00:00 2001
From: Matthew Murray <matthewmurray711@gmail.com>
Date: Wed, 26 Aug 2026 13:51:18 +0000
Subject: [PATCH] Default pinned host memory to enabled for cudf-polars' Ray,
 Dask, and SPMD engines

---
 .../cudf_polars/cudf_polars/engine/options.py | 23 +++++++++++--------
 .../tests/streaming/test_options.py           | 19 +++++++++++----
 2 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/python/cudf_polars/cudf_polars/engine/options.py b/python/cudf_polars/cudf_polars/engine/options.py
index b519d2aba9d1..ddd22e6be739 100644
--- a/python/cudf_polars/cudf_polars/engine/options.py
+++ b/python/cudf_polars/cudf_polars/engine/options.py
@@ -45,6 +45,7 @@ def _opt(
     category: str,
     env_var: str | None = None,
     coerce: Callable[[str], Any] = str,
+    default: Any = UNSPECIFIED,
 ) -> Any:
     """
     Factory for ``StreamingOptions`` fields with category and env-var metadata.
@@ -59,10 +60,14 @@ def _opt(
         :class:`StreamingOptions` is instantiated without an explicit value for
         this field, the factory reads the environment variable (if set) on the constructing
         process.  ``None`` means no environment variable; the field defaults to
-        :data:`UNSPECIFIED`.
+        *default*.
     coerce
         Callable used to convert the raw env-var string to the field's type.
         Defaults to ``str`` (no conversion).
+    default
+        Value used when neither an explicit value nor the environment variable
+        is set. Defaults to :data:`UNSPECIFIED`, which defers to rapidsmpf's
+        built-in default.
     """
 
     def _default() -> Any:
@@ -70,7 +75,7 @@ def _default() -> Any:
             raw = os.environ.get(env_var)
             if raw is not None:
                 return coerce(raw)
-        return UNSPECIFIED
+        return default
 
     return dataclasses.field(
         default_factory=_default,
@@ -160,7 +165,7 @@ class StreamingOptions:
     pinned_memory
         Enable pinned host memory.
         Env: ``RAPIDSMPF_PINNED_MEMORY``.
-        Default: ``False``.
+        Default: ``True``.
         Category: rapidsmpf.
     pinned_initial_pool_size
         Initial pinned memory pool size (bytes).
@@ -308,10 +313,10 @@ class StreamingOptions:
         "rapidsmpf", "RAPIDSMPF_ALLOW_OVERBOOKING_BY_DEFAULT", parse_boolean
     )
     pinned_memory: bool | Unspecified = _opt(
-        "rapidsmpf", "RAPIDSMPF_PINNED_MEMORY", parse_boolean
+        "rapidsmpf", "RAPIDSMPF_PINNED_MEMORY", parse_boolean, default=True
     )
     pinned_initial_pool_size: int | Unspecified = _opt(
-        "rapidsmpf", "RAPIDSMPF_PINNED_INITIAL_POOL_SIZE", int
+        "rapidsmpf", "RAPIDSMPF_PINNED_INITIAL_POOL_SIZE", int, default=0
     )
     pinned_max_pool_size: str | Unspecified = _opt(
         "rapidsmpf", "RAPIDSMPF_PINNED_MAX_POOL_SIZE"
@@ -427,8 +432,8 @@ def to_dict(self) -> dict[str, Any]:
 
         Examples
         --------
-        >>> StreamingOptions(fallback_mode="silent").to_dict()
-        {'fallback_mode': 'silent'}
+        >>> StreamingOptions(fallback_mode="silent").to_dict()  # doctest: +ELLIPSIS
+        {..., 'fallback_mode': 'silent'}
         >>> StreamingOptions.from_dict(
         ...     StreamingOptions(fallback_mode="silent").to_dict()
         ... )  # doctest: +ELLIPSIS
@@ -633,7 +638,7 @@ def _add_cli_args(parser: argparse.ArgumentParser) -> None:
             action=argparse.BooleanOptionalAction,
             help=textwrap.dedent("""\
                 Enable pinned host memory if available on the system.
-                Env: RAPIDSMPF_PINNED_MEMORY. Built-in default: false."""),
+                Env: RAPIDSMPF_PINNED_MEMORY. Default: true."""),
         )
         g.add_argument(
             "--pinned-initial-pool-size",
@@ -642,7 +647,7 @@ def _add_cli_args(parser: argparse.ArgumentParser) -> None:
             type=int,
             help=textwrap.dedent("""\
                 Starting allocation for the pinned memory pool in bytes.
-                Env: RAPIDSMPF_PINNED_INITIAL_POOL_SIZE. Built-in default: 0."""),
+                Env: RAPIDSMPF_PINNED_INITIAL_POOL_SIZE. Default: 0."""),
         )
         g.add_argument(
             "--pinned-max-pool-size",
diff --git a/python/cudf_polars/tests/streaming/test_options.py b/python/cudf_polars/tests/streaming/test_options.py
index f0adf211f72c..dece05cf2eca 100644
--- a/python/cudf_polars/tests/streaming/test_options.py
+++ b/python/cudf_polars/tests/streaming/test_options.py
@@ -143,7 +143,10 @@ def test_rapidsmpf_options_unspecified_fields_absent() -> None:
         k: os.environ.pop(k) for k in list(os.environ) if k.startswith("RAPIDSMPF_")
     }
     try:
-        assert StreamingOptions().to_rapidsmpf_options().get_strings() == {}
+        assert StreamingOptions().to_rapidsmpf_options().get_strings() == {
+            "pinned_memory": "True",
+            "pinned_initial_pool_size": "0",
+        }
     finally:
         os.environ.update(env_backup)
 
@@ -385,14 +388,22 @@ def test_from_argparse_omitted_flag_still_picks_up_env_var(
 # ---------------------------------------------------------------------------
 
 
-def test_to_dict_empty_when_all_unspecified() -> None:
-    assert StreamingOptions().to_dict() == {}
+def test_to_dict_only_defaulted_fields_when_all_unspecified() -> None:
+    assert StreamingOptions().to_dict() == {
+        "pinned_memory": True,
+        "pinned_initial_pool_size": 0,
+    }
 
 
 def test_to_dict_contains_only_set_fields() -> None:
     opts = StreamingOptions(fallback_mode="silent", num_streaming_threads=4)
     d = opts.to_dict()
-    assert d == {"fallback_mode": "silent", "num_streaming_threads": 4}
+    assert d == {
+        "fallback_mode": "silent",
+        "num_streaming_threads": 4,
+        "pinned_memory": True,
+        "pinned_initial_pool_size": 0,
+    }
 
 
 def test_to_dict_roundtrip() -> None: