Run cudf_polars unit tests with RapidsMPF

ci/run_cudf_polars_with_rapidsmpf_pytests.sh

@@ -14,10 +14,11 @@ cd "$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/../python/cudf_polars/
 
 # Run experimental tests with the "single" cluster mode and the "rapidsmpf" runtime
 rapids-logger "Running experimental tests with the 'rapidsmpf' runtime and a 'single' cluster"
-timeout 10m python -m pytest --cache-clear "$@" "tests/experimental" \
+CUDF_POLARS__PARQUET_OPTIONS__USE_RAPIDSMPF_NATIVE=1 CUDF_POLARS__EXECUTOR__SHUFFLE_METHOD=rapidsmpf timeout 10m python -m pytest --cache-clear "$@" tests \
     --executor streaming \
     --cluster single \
-    --runtime rapidsmpf
+    --runtime rapidsmpf \
+    --blocksize-mode small
 
 # Run experimental tests with the "distributed" cluster mode and the "rapidsmpf" runtime
 rapids-logger "Running experimental tests with the 'rapidsmpf' runtime and a 'distributed' cluster"

dependencies.yaml

@@ -115,6 +115,8 @@ files:
       - depends_on_cudf_kafka
       - depends_on_custreamz
       - depends_on_cudf_polars
+      - depends_on_rapidsmpf
+      - cudf_polars_trace
   test_java:
     output: none
     includes:

python/cudf_polars/cudf_polars/experimental/dask_registers.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 
 """Dask function registrations such as serializers and dispatch implementations."""
@@ -215,7 +215,7 @@ def _(x: DataFrame) -> int:
         from rapidsmpf.integrations.dask.spilling import register_dask_serialize
 
         register_dask_serialize()  # pragma: no cover; rapidsmpf dependency not included yet
-    except ImportError:
+    except ImportError:  # pragma: no cover
         pass
 
     # Register the tokenizer for NamedExpr and DataType. This is a performance

python/cudf_polars/tests/conftest.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
@@ -28,6 +28,11 @@ def clear_memory_resource_cache():
     cudf_polars.callback.default_memory_resource.cache_clear()
 
 
+@pytest.fixture
+def using_rapidsmpf():
+    return cudf_polars.testing.asserts.DEFAULT_RUNTIME == "rapidsmpf"
+
+
 def pytest_addoption(parser):
     parser.addoption(
         "--executor",

python/cudf_polars/tests/experimental/rapidsmpf/test_tracing.py

@@ -20,6 +20,7 @@
 @pytest.mark.skipif(DEFAULT_CLUSTER != "single", reason="Requires 'single' cluster.")
 def test_structlog_streaming_node_events():
     """Test that structlog emits 'Streaming Actor' events when tracing is enabled."""
+    pytest.importorskip("structlog")
     # Run in subprocess to control CUDF_POLARS_LOG_TRACES environment variable
     code = textwrap.dedent("""\
     import polars as pl

python/cudf_polars/tests/expressions/test_agg.py

@@ -194,7 +194,13 @@ def test_implode_agg_unsupported():
     assert_ir_translation_raises(q, NotImplementedError)
 
 
-def test_decimal_aggs(decimal_df: pl.LazyFrame) -> None:
+def test_decimal_aggs(decimal_df: pl.LazyFrame, using_rapidsmpf, request) -> None:
+    request.applymarker(
+        pytest.mark.xfail(
+            condition=using_rapidsmpf,
+            reason="https://github.com/rapidsai/cudf/issues/21645",
+        )
+    )
     q = decimal_df.with_columns(
         sum=pl.col("a").sum(),
         min=pl.col("a").min(),

python/cudf_polars/tests/expressions/test_booleanfunction.py

@@ -76,8 +76,14 @@ def test_booleanfunction_all_any_kleene(expr, ignore_nulls):
 )
 @pytest.mark.parametrize("has_nans", [False, True], ids=["no_nans", "nans"])
 def test_boolean_function_unary(
-    expr: Callable[[pl.Expr], pl.Expr], *, has_nans: bool, has_nulls: bool
+    expr: Callable[[pl.Expr], pl.Expr],
+    *,
+    has_nans: bool,
+    has_nulls: bool,
+    using_rapidsmpf: bool,
 ) -> None:
+    if using_rapidsmpf:
+        pytest.skip("Can segault on 12.9 builds with rapidsmpf runtime.")
     values: list[float | None] = [1, 2, 3, 4, 5]
     if has_nans:
         values[3] = float("nan")
@@ -160,7 +166,9 @@ def test_boolean_isbetween(closed, bounds):
     "expr", [pl.any_horizontal("*"), pl.all_horizontal("*")], ids=["any", "all"]
 )
 @pytest.mark.parametrize("wide", [False, True], ids=["narrow", "wide"])
-def test_boolean_horizontal(expr, has_nulls, wide):
+def test_boolean_horizontal(expr, has_nulls, wide, using_rapidsmpf):
+    if using_rapidsmpf:
+        pytest.skip("Can segault on 12.9 builds with rapidsmpf runtime.")
     ldf = pl.LazyFrame(
         {
             "a": [False, False, False, False, False, True],

python/cudf_polars/tests/expressions/test_len.py

@@ -30,7 +30,13 @@ def test_len(dtype, empty):
 
 
 @pytest.mark.parametrize("data", [[1, 2, 3], [1, 2, None]])
-def test_col_len(data):
+def test_col_len(data, using_rapidsmpf, request):
+    request.applymarker(
+        pytest.mark.xfail(
+            condition=using_rapidsmpf,
+            reason="https://github.com/rapidsai/cudf/issues/21645",
+        )
+    )
     data = {"a": list("xyz"), "b": data}
     q = pl.LazyFrame(data).select(
         pl.col("a").len().alias("l"),

python/cudf_polars/tests/expressions/test_stringfunction.py

@@ -659,7 +659,9 @@ def test_string_zfill_forbidden_chars():
         ),
     ],
 )
-def test_string_pad_start(width, char):
+def test_string_pad_start(width, char, using_rapidsmpf):
+    if using_rapidsmpf:
+        pytest.skip("Can segault on 12.9 builds with rapidsmpf runtime.")
     df = pl.LazyFrame({"a": ["abc", "defg", "hij"]})
     q = df.select(pl.col("a").str.pad_start(width, char))
     assert_gpu_result_equal(q)

python/cudf_polars/tests/test_config.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 from typing import Any, cast
 
@@ -381,8 +382,12 @@ def test_scheduler_deprecated() -> None:
 
 def test_validate_shuffle_method_defaults(
     *,
+    monkeypatch: pytest.MonkeyPatch,
     rapidsmpf_distributed_available: bool,
 ) -> None:
+    # TODO: Can remove once removed from ci/run_cudf_polars_pytests.sh
+    # and the default shuffle method changes to "rapidsmpf"
+    monkeypatch.delenv("CUDF_POLARS__EXECUTOR__SHUFFLE_METHOD", raising=False)
     config = ConfigOptions.from_polars_engine(
         pl.GPUEngine(
             executor="streaming",
@@ -804,7 +809,12 @@ def test_cuda_stream_policy_from_env(
     elif runtime == "rapidsmpf":
         with pytest.raises(ValueError, match="The rapidsmpf streaming engine"):
             ConfigOptions.from_polars_engine(engine)
-    else:
+    elif (
+        os.environ.get("CUDF_POLARS__EXECUTOR__SHUFFLE_METHOD", "") != "rapidsmpf"
+        and not rapidsmpf_single_available
+    ):
+        # CUDF_POLARS__EXECUTOR__SHUFFLE_METHOD set in ci/run_cudf_polars_pytests.sh
+        # Would raise since rapidsmpf shuffle requires rapidsmpf to be installed
         config = ConfigOptions.from_polars_engine(engine)
         assert config.cuda_stream_policy == env
 

python/cudf_polars/tests/test_dataframescan.py

@@ -86,7 +86,13 @@ def test_dataframescan_with_decimals():
     POLARS_VERSION_LT_138,
     reason="height parameter added in Polars 1.38",
 )
-def test_dataframescan_zero_width_with_rows():
+def test_dataframescan_zero_width_with_rows(request, using_rapidsmpf):
+    request.applymarker(
+        pytest.mark.xfail(
+            using_rapidsmpf,
+            reason="https://github.com/rapidsai/cudf/issues/21644",
+        )
+    )
     df = pl.LazyFrame(height=5)
     q = df.select(pl.len())
     assert_gpu_result_equal(q)

python/cudf_polars/tests/test_drop_nulls.py

@@ -42,7 +42,15 @@ def test_drop_null(null_data):
     [0, pl.col("a").mean(), pl.col("b")],
     ids=["scalar", "aggregation", "column_expression"],
 )
-def test_fill_null(null_data, value):
+def test_fill_null(null_data, value, using_rapidsmpf, request):
+    request.applymarker(
+        pytest.mark.xfail(
+            condition=using_rapidsmpf
+            and "mean" in str(value)
+            and len(null_data.collect()) == 2,
+            reason="https://github.com/rapidsai/cudf/issues/21721",
+        )
+    )
     q = null_data.select(pl.col("a").fill_null(value))
     assert_gpu_result_equal(q)
 

python/cudf_polars/tests/test_groupby.py

@@ -126,7 +126,33 @@ def maintain_order(request):
     return request.param
 
 
-def test_groupby(df: pl.LazyFrame, maintain_order, keys, exprs):
+def test_groupby(
+    df: pl.LazyFrame, maintain_order, keys, exprs, using_rapidsmpf, request
+):
+    if using_rapidsmpf:
+        pytest.skip("Can segault on 12.9 builds with rapidsmpf runtime.")
+    failing_rapidsmpf_nodeids = {
+        'test_groupby[maintain_order-col("key1")-col("int32").mean()]',
+        'test_groupby[maintain_order-col("key1")-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-col("key2")-col("int32").mean()]',
+        'test_groupby[maintain_order-col("key2")-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-col("key1")-dyn int: 1-col("int32").mean()]',
+        'test_groupby[maintain_order-col("key1")-dyn int: 1-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-[(col("key1")) * (col("key2"))]-col("int32").mean()]',
+        'test_groupby[maintain_order-[(col("key1")) * (col("key2"))]-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-col("key1")-col("key2")-col("int32").mean()]',
+        'test_groupby[maintain_order-col("key1")-col("key2")-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-[(col("key1")) == (col("key2"))]-col("int32").mean()]',
+        'test_groupby[maintain_order-[(col("key1")) == (col("key2"))]-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+        'test_groupby[maintain_order-col("key2")-[(col("key1")) == (1)]-col("int32").mean()]',
+        'test_groupby[maintain_order-col("key2")-[(col("key1")) == (1)]-col("uint16_with_null").sum()-col("uint16_with_null").mean().alias("mean")]',
+    }
+    request.applymarker(
+        pytest.mark.xfail(
+            using_rapidsmpf and request.node.name in failing_rapidsmpf_nodeids,
+            reason="https://github.com/rapidsai/cudf/issues/21642 and probably mean calculation referenced in https://github.com/rapidsai/cudf/issues/21721",
+        )
+    )
     q = df.group_by(*keys, maintain_order=maintain_order).agg(*exprs)
 
     if not maintain_order:
@@ -136,7 +162,14 @@ def test_groupby(df: pl.LazyFrame, maintain_order, keys, exprs):
     assert_gpu_result_equal(q, check_exact=False)
 
 
-def test_groupby_sorted_keys(df: pl.LazyFrame, keys, exprs):
+def test_groupby_sorted_keys(df: pl.LazyFrame, keys, exprs, using_rapidsmpf, request):
+    request.applymarker(
+        pytest.mark.xfail(
+            using_rapidsmpf,
+            strict=False,
+            reason="https://github.com/rapidsai/cudf/issues/21642 -  no deterministic sort for keys",
+        )
+    )
     sorted_keys = [
         key.sort(descending=descending)
         for key, descending in zip(keys, itertools.cycle([False, True]))

python/cudf_polars/tests/test_join.py

@@ -70,7 +70,16 @@ def test_join_maintain_order(left, right, maintain_order):
         ["c", "a"],
     ],
 )
-def test_non_coalesce_join(left, right, how, nulls_equal, join_expr):
+def test_non_coalesce_join(
+    left, right, how, nulls_equal, join_expr, using_rapidsmpf, request
+):
+    request.applymarker(
+        pytest.mark.xfail(
+            using_rapidsmpf,
+            strict=False,
+            reason="Non deterministic sort/join on nulls",
+        )
+    )
     query = left.join(
         right, on=join_expr, how=how, nulls_equal=nulls_equal, coalesce=False
     )

python/cudf_polars/tests/test_merge_sorted.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES.
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
@@ -9,17 +9,14 @@
 from cudf_polars.testing.asserts import assert_gpu_result_equal
 
 
-@pytest.mark.parametrize(
-    "descending",
-    [
-        pytest.param(
-            True,
-            marks=pytest.mark.xfail(reason="polars/issues/21511"),
-        ),
-        False,
-    ],
-)
-def test_merge_sorted_without_nulls(descending):
+@pytest.mark.parametrize("descending", [True, False])
+def test_merge_sorted_without_nulls(descending, request, using_rapidsmpf):
+    request.applymarker(
+        pytest.mark.xfail(
+            not using_rapidsmpf and descending,
+            reason="https://github.com/pola-rs/polars/issues/21511",
+        )
+    )
     df0 = pl.LazyFrame(
         {"name": ["steve", "elise", "bob"], "age": [42, 44, 18], "height": [5, 6, 5]}
     ).sort("age", descending=descending)

python/cudf_polars/tests/test_rolling.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import contextlib
 from datetime import datetime
 
 import pytest
@@ -133,7 +134,7 @@ def test_grouped_rolling():
     assert_gpu_result_equal(q)
 
 
-def test_grouped_rolling_unsorted_raises():
+def test_grouped_rolling_unsorted_raises(using_rapidsmpf):
     df = pl.LazyFrame(
         {
             "keys": [1, None, 2, 1, 2, None],
@@ -145,7 +146,19 @@ def test_grouped_rolling_unsorted_raises():
 
     with pytest.raises(pl.exceptions.ComputeError):
         q.collect(engine="in-memory")
-    with pytest.raises(RuntimeError, match="Input for grouped rolling is not sorted"):
+
+    fallback_warning = (
+        pytest.warns(
+            UserWarning,
+            match="shuffle_method=rapidsmpf-single does not support maintain_order=True",
+        )
+        if using_rapidsmpf
+        else contextlib.nullcontext()
+    )
+    with (
+        pytest.raises(RuntimeError, match="Input for grouped rolling is not sorted"),
+        fallback_warning,
+    ):
         q.collect(engine=pl.GPUEngine(raise_on_fail=True))
 
 

python/cudf_polars/tests/test_scan.py

@@ -702,7 +702,13 @@ def test_scan_tiny_file_not_compressed(tmp_path):
     reason="height parameter added in Polars 1.38",
 )
 @pytest.mark.parametrize("engine", [None, NO_CHUNK_ENGINE])
-def test_scan_parquet_zero_width_with_limit(tmp_path, engine):
+def test_scan_parquet_zero_width_with_limit(tmp_path, engine, request, using_rapidsmpf):
+    request.applymarker(
+        pytest.mark.xfail(
+            using_rapidsmpf and engine is None,
+            reason="https://github.com/rapidsai/cudf/issues/21644",
+        )
+    )
     path = tmp_path / "zero_width.parquet"
     pl.LazyFrame(height=20).sink_parquet(path)
     q = pl.scan_parquet(path).head(5)

python/cudf_polars/tests/test_select.py

@@ -87,7 +87,13 @@ def test_select_with_cse_no_agg():
     assert_gpu_result_equal(query)
 
 
-def test_select_with_cse_with_agg():
+def test_select_with_cse_with_agg(using_rapidsmpf, request):
+    request.applymarker(
+        pytest.mark.xfail(
+            condition=using_rapidsmpf,
+            reason="https://github.com/rapidsai/cudf/issues/21645",
+        )
+    )
     df = pl.LazyFrame({"a": [1, 2, 3]})
     expr = pl.col("a") + pl.col("a")
     asum = pl.col("a").sum() + pl.col("a").sum()

python/cudf_polars/tests/test_tracing.py

@@ -111,9 +111,9 @@ def test_log_query_plan() -> None:
         [sys.executable, "-c", code], env=env, stderr=subprocess.STDOUT
     )
 
-    # Check for Query Plan event
+    # Check for Query Plan event generated from SerializablePlan
     assert b"Query Plan" in result
     assert b"scope=plan" in result or b"'scope': 'plan'" in result
-    assert b"ir_id" in result
-    assert b"ir_type" in result
-    assert b"children_ir_ids" in result
+    assert b"actor_ir_id" in result
+    assert b"actor_ir_type" in result
+    assert b"children" in result