[Multi-GPU Polars] Introduce Ray mode for multi-GPU cudf-polars execution

python/cudf_polars/cudf_polars/experimental/benchmarks/utils.py

@@ -33,7 +33,7 @@
 
 import rmm.statistics
 
-from cudf_polars.experimental.rapidsmpf.spmd import spmd_execution
+from cudf_polars.experimental.rapidsmpf.frontend.spmd import spmd_execution
 
 # The dtype for count() aggregations depends on the presence
 # of the polars-runtime-64 package (`polars[rt64]`).
@@ -1866,7 +1866,9 @@ def run_polars_spmd(
         cuda_stream_policy=run_config.stream_policy,
     ) as (comm, ctx, engine):
         from cudf_polars.experimental.rapidsmpf.collectives.common import reserve_op_id
-        from cudf_polars.experimental.rapidsmpf.spmd import allgather_polars_dataframe
+        from cudf_polars.experimental.rapidsmpf.frontend.spmd import (
+            allgather_polars_dataframe,
+        )
 
         def _allgather_result(df: pl.DataFrame) -> pl.DataFrame:
             with reserve_op_id() as op_id:

python/cudf_polars/cudf_polars/experimental/rapidsmpf/collectives/__init__.py

@@ -1,9 +1,12 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 """Collective operations for the RapidsMPF streaming runtime."""
 
 from __future__ import annotations
 
-from cudf_polars.experimental.rapidsmpf.collectives.common import ReserveOpIDs
+from cudf_polars.experimental.rapidsmpf.collectives.common import (
+    ReserveOpIDs,
+    reserve_op_id,
+)
 
-__all__ = ["ReserveOpIDs"]
+__all__ = ["ReserveOpIDs", "reserve_op_id"]

python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py

@@ -108,47 +108,62 @@ def evaluate_logical_plan(
         # Build and execute the streaming pipeline.
         # This must be done on all worker processes
         # for cluster == "distributed".
-        if (
-            config_options.executor.cluster == "distributed"
-        ):  # pragma: no cover; block depends on executor type and Distributed cluster
-            # Distributed execution: Use client.run
-
-            # NOTE: Distributed execution requires Dask for now
-            from cudf_polars.experimental.rapidsmpf.dask import evaluate_pipeline_dask
-
-            result, metadata_collector = evaluate_pipeline_dask(
-                evaluate_pipeline,
-                ir,
-                partition_info,
-                config_options,
-                stats,
-                collective_id_map,
-                collect_metadata=collect_metadata,
-            )
-        elif config_options.executor.cluster == "spmd":
-            from cudf_polars.experimental.rapidsmpf.spmd import (
-                evaluate_pipeline_spmd_mode,
-            )
+        match config_options.executor.cluster:
+            case "distributed":  # pragma: no cover; block depends on executor type and Distributed cluster
+                # Distributed execution: Use client.run
+                # NOTE: Distributed execution requires Dask for now
+                from cudf_polars.experimental.rapidsmpf.dask import (
+                    evaluate_pipeline_dask,
+                )
 
-            result, metadata_collector = evaluate_pipeline_spmd_mode(
-                ir,
-                partition_info,
-                config_options,
-                stats,
-                collective_id_map,
-                collect_metadata=collect_metadata,
-            )
-        else:
-            # Single-process execution: Run locally
-            result, metadata_collector = evaluate_pipeline(
-                ir,
-                partition_info,
-                config_options,
-                stats,
-                collective_id_map,
-                single_process_communicator(Options(), ProgressThread()),
-                collect_metadata=collect_metadata,
-            )
+                result, metadata_collector = evaluate_pipeline_dask(
+                    evaluate_pipeline,
+                    ir,
+                    partition_info,
+                    config_options,
+                    stats,
+                    collective_id_map,
+                    collect_metadata=collect_metadata,
+                )
+            case "spmd":
+                from cudf_polars.experimental.rapidsmpf.frontend.spmd import (
+                    evaluate_pipeline_spmd_mode,
+                )
+
+                result, metadata_collector = evaluate_pipeline_spmd_mode(
+                    ir,
+                    partition_info,
+                    config_options,
+                    stats,
+                    collective_id_map,
+                    collect_metadata=collect_metadata,
+                )
+            case "ray":
+                from cudf_polars.experimental.rapidsmpf.frontend.ray import (
+                    evaluate_pipeline_ray_mode,
+                )
+
+                result, metadata_collector = evaluate_pipeline_ray_mode(
+                    ir,
+                    partition_info,
+                    config_options,
+                    stats,
+                    collective_id_map,
+                    collect_metadata=collect_metadata,
+                )
+            case "single":
+                # Single-process execution: Run locally
+                result, metadata_collector = evaluate_pipeline(
+                    ir,
+                    partition_info,
+                    config_options,
+                    stats,
+                    collective_id_map,
+                    single_process_communicator(Options(), ProgressThread()),
+                    collect_metadata=collect_metadata,
+                )
+            case other:
+                raise ValueError(f"Unknown cluster mode: {other}")
 
     return result, metadata_collector
 

python/cudf_polars/cudf_polars/experimental/rapidsmpf/frontend/__init__.py

@@ -0,0 +1,7 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+# SPDX-License-Identifier: Apache-2.0
+"""Multi-GPU frontend execution modes for the RapidsMPF streaming engine."""
+
+from __future__ import annotations
+
+__all__: list[str] = []