Consolidate evaluate_rapidsmpf into evaluate_streaming in cudf_polars

dependencies.yaml

@@ -50,50 +50,6 @@ files:
       - test_python_cudf_pandas
       - test_python_cudf_polars
       - test_python_s3
-  devcontainers:
-    output: none
-    # Same as "all", excluding depends_on_rapidsmpf (which is built from source)
-    includes:
-      - build_base
-      - build_all
-      - build_cpp
-      - build_python_common
-      - clang
-      - cuda
-      - cuda_version
-      - cudf_polars_trace
-      - depends_on_cupy
-      - depends_on_cuda_python
-      - depends_on_dask_cuda
-      - depends_on_libkvikio
-      - depends_on_librmm
-      - depends_on_libnvcomp
-      - depends_on_numba_cuda
-      - depends_on_rapids_logger
-      - depends_on_rmm
-      - develop
-      - docs
-      - iwyu
-      - notebooks
-      - numpy_run
-      - py_version
-      - pyarrow_run
-      - rapids_build_skbuild
-      - rapids_build_setuptools
-      - run_common
-      - run_cudf
-      - run_cudf_polars
-      - run_pylibcudf
-      - run_dask_cudf
-      - run_custreamz
-      - test_cpp
-      - test_python_common
-      - test_python_cudf
-      - test_python_cudf_common
-      - test_python_pylibcudf
-      - test_python_cudf_pandas
-      - test_python_cudf_polars
-      - test_python_s3
   test_static_build:
     output: none
     includes:

python/cudf_polars/cudf_polars/experimental/expressions.py

@@ -125,10 +125,7 @@ def select(
         A mapping from unique nodes in the new graph to associated
         partitioning information.
     """
-    output_names = [next(names) for _ in range(len(exprs))]
-    named_exprs = [
-        NamedExpr(name, expr) for name, expr in zip(output_names, exprs, strict=True)
-    ]
+    named_exprs = [NamedExpr(next(names), expr) for expr in exprs]
     new_ir: IR = Select(
         {ne.name: ne.value.dtype for ne in named_exprs},
         named_exprs,

python/cudf_polars/cudf_polars/experimental/parallel.py

@@ -101,30 +101,6 @@ def lower_ir_graph(
     return mapper(ir)
 
 
-def evaluate_rapidsmpf(
-    ir: IR,
-    config_options: ConfigOptions[StreamingExecutor],
-) -> pl.DataFrame:  # pragma: no cover; rapidsmpf runtime not tested in CI yet
-    """
-    Evaluate with the RapidsMPF streaming runtime.
-
-    Parameters
-    ----------
-    ir
-        Logical plan to evaluate.
-    config_options
-        GPUEngine configuration options.
-
-    Returns
-    -------
-    A cudf-polars DataFrame object.
-    """
-    from cudf_polars.experimental.rapidsmpf.core import evaluate_logical_plan
-
-    result, _ = evaluate_logical_plan(ir, config_options, collect_metadata=False)
-    return result
-
-
 def evaluate_streaming(
     ir: IR,
     config_options: ConfigOptions[StreamingExecutor],
@@ -146,7 +122,10 @@ def evaluate_streaming(
     # Clear source info cache in case data was overwritten
     _clear_source_info_cache()
 
-    return evaluate_rapidsmpf(ir, config_options)
+    from cudf_polars.experimental.rapidsmpf.core import evaluate_logical_plan
+
+    result, _ = evaluate_logical_plan(ir, config_options, collect_metadata=False)
+    return result
 
 
 @lower_ir_node.register(Union)