Default pinned host memory to enabled for cudf-polars' Ray, Dask, and SPMD engines

python/cudf_polars/cudf_polars/engine/core.py

@@ -211,6 +211,8 @@ def resolve_rapidsmpf_options(rapidsmpf_options: Options | None) -> Options:
 
     - ``num_streaming_threads=4``: moderate worker count for the rapidsmpf
       streaming runtime, shared across frontends.
+    - ``pinned_memory=true``, ``pinned_initial_pool_size=0``: pinned host
+      memory enabled by default.
 
     Parameters
     ----------
@@ -226,7 +228,13 @@ def resolve_rapidsmpf_options(rapidsmpf_options: Options | None) -> Options:
     if rapidsmpf_options is None:
         rapidsmpf_options = Options(get_environment_variables())
 
-    rapidsmpf_options.insert_if_absent({"num_streaming_threads": "4"})
+    rapidsmpf_options.insert_if_absent(
+        {
+            "num_streaming_threads": "4",
+            "pinned_memory": "true",
+            "pinned_initial_pool_size": "0",
+        }
+    )
     return rapidsmpf_options
 
 

python/cudf_polars/cudf_polars/engine/options.py

@@ -46,6 +46,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.
@@ -60,18 +61,22 @@ 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:
         if env_var:
             raw = os.environ.get(env_var)
             if raw is not None:
                 return coerce(raw)
-        return UNSPECIFIED
+        return default
 
     return dataclasses.field(
         default_factory=_default,
@@ -161,7 +166,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).
@@ -433,8 +438,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
@@ -639,7 +644,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",
@@ -648,7 +653,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",

python/cudf_polars/tests/streaming/test_options.py

@@ -418,10 +418,6 @@ 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_contains_only_set_fields() -> None:
     opts = StreamingOptions(
         fallback_mode="silent",