Run Polars TPC benchmarks in CI

.github/workflows/pr.yaml

@@ -41,6 +41,7 @@ jobs:
       - wheel-build-cudf-polars
       - wheel-tests-cudf-polars
       - cudf-polars-polars-tests
+      - tpc-tests-cudf-polars
       - wheel-build-dask-cudf
       - wheel-tests-dask-cudf
       - devcontainer
@@ -816,6 +817,23 @@ jobs:
       matrix_filter: map(select(.ARCH == "amd64")) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))]))
       build_type: pull-request
       script: "ci/test_cudf_polars_polars_tests.sh"
+  tpc-tests-cudf-polars:
+    needs: [wheel-build-cudf-polars, wheel-build-cudf-streaming, changed-files]
+    permissions:
+      actions: read
+      contents: read
+      id-token: write
+      packages: read
+      pull-requests: read
+    secrets: inherit # zizmor: ignore[secrets-inherit]
+    uses: rapidsai/shared-workflows/.github/workflows/wheels-test.yaml@main
+    if: fromJSON(needs.changed-files.outputs.changed_file_groups).test_python_wheels && fromJSON(needs.changed-files.outputs.changed_file_groups).neither_cudf_nor_dask_cudf
+    with:
+      # This selects "ARCH=amd64 + the latest supported Python + CUDA".
+      matrix_filter: map(select(.ARCH == "amd64")) | group_by(.CUDA_VER|split(".")|map(tonumber)|.[0]) | map(max_by([(.PY_VER|split(".")|map(tonumber)), (.CUDA_VER|split(".")|map(tonumber))]))
+      build_type: pull-request
+      container-options: "--cap-add CAP_SYS_PTRACE --shm-size=8g --ulimit=nofile=1000000:1000000"
+      script: "ci/run_cudf_polars_tpc.sh"
   wheel-build-dask-cudf:
     needs: [build-details, wheel-build-cudf]
     permissions:

ci/run_cudf_polars_pytests.sh

@@ -11,4 +11,6 @@ TIMEOUT_TOOL_PATH="$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/timeout_with_st
 cd "$(dirname "$(realpath "${BASH_SOURCE[0]}")")"/../python/cudf_polars/
 
 python "${TIMEOUT_TOOL_PATH}" --enable-python 5400 \
-       python -m pytest --cache-clear -p no:benchmark "$@" tests
+       python -m pytest --cache-clear -p no:benchmark "$@" tests \
+           --ignore=tests/streaming/test_tpch.py \
+           --ignore=tests/streaming/test_tpcds.py

ci/run_cudf_polars_tpc.sh

@@ -0,0 +1,50 @@
+#!/bin/bash
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+set -euo pipefail
+
+source rapids-init-pip
+
+rapids-logger "Download wheels"
+
+RAPIDS_PY_CUDA_SUFFIX="$(rapids-wheel-ctk-name-gen "${RAPIDS_CUDA_VERSION}")"
+LIBCUDF_WHEELHOUSE=$(rapids-download-from-github "$(rapids-artifact-name wheel_cpp libcudf cudf --cuda "$RAPIDS_CUDA_VERSION")")
+PYLIBCUDF_WHEELHOUSE=$(rapids-download-from-github "$(rapids-artifact-name wheel_python pylibcudf cudf --stable --cuda "$RAPIDS_CUDA_VERSION")")
+CUDF_POLARS_WHEELHOUSE=$(rapids-download-from-github "$(rapids-artifact-name wheel_python cudf-polars cudf --pure --arch any --cuda "$RAPIDS_CUDA_VERSION")")
+LIBCUDF_STREAMING_WHEELHOUSE=$(rapids-download-from-github "$(rapids-artifact-name wheel_cpp libcudf-streaming cudf --cuda "$RAPIDS_CUDA_VERSION")")
+CUDF_STREAMING_WHEELHOUSE=$(rapids-download-from-github "$(rapids-artifact-name wheel_python cudf-streaming cudf --stable --cuda "$RAPIDS_CUDA_VERSION")")
+
+rapids-generate-pip-constraints py_test_cudf_polars "${PIP_CONSTRAINT}"
+
+rapids-logger "Installing cudf_polars and TPC test dependencies"
+
+TPCH_REQUIREMENTS=$(mktemp --suffix=.txt)
+rapids-dependency-file-generator \
+    --config dependencies.yaml \
+    --file-key test_cudf_polars_tpch \
+    --output requirements \
+    --matrix "cuda=${RAPIDS_CUDA_VERSION%.*};arch=$(arch);py=${RAPIDS_PY_VERSION}" \
+    > "${TPCH_REQUIREMENTS}"
+
+rapids-pip-retry install \
+    -v \
+    --prefer-binary \
+    --constraint "${PIP_CONSTRAINT}" \
+    "$(echo "${CUDF_POLARS_WHEELHOUSE}"/cudf_polars_"${RAPIDS_PY_CUDA_SUFFIX}"*.whl)[test]" \
+    "$(echo "${LIBCUDF_WHEELHOUSE}"/libcudf_"${RAPIDS_PY_CUDA_SUFFIX}"*.whl)" \
+    "$(echo "${PYLIBCUDF_WHEELHOUSE}"/pylibcudf_"${RAPIDS_PY_CUDA_SUFFIX}"*.whl)" \
+    "$(echo "${LIBCUDF_STREAMING_WHEELHOUSE}"/libcudf_streaming_"${RAPIDS_PY_CUDA_SUFFIX}"*.whl)" \
+    "$(echo "${CUDF_STREAMING_WHEELHOUSE}"/cudf_streaming_"${RAPIDS_PY_CUDA_SUFFIX}"*.whl)" \
+    -r "${TPCH_REQUIREMENTS}"
+
+rapids-logger "Check GPU usage"
+nvidia-smi
+
+rapids-logger "Running TPC-H and TPC-DS validation tests"
+
+cd python/cudf_polars
+
+python -m pytest tests/streaming/test_tpch.py tests/streaming/test_tpcds.py \
+    --iterations 2 \
+    -v

dependencies.yaml

@@ -475,6 +475,14 @@ files:
       - test_python_common
       - test_python_cudf_polars
       - cudf_polars_trace
+  py_tpch_cudf_polars:
+    output: pyproject
+    pyproject_dir: python/cudf_polars
+    extras:
+      table: project.optional-dependencies
+      key: tpch
+    includes:
+      - test_cudf_polars_tpch
   py_trace_cudf_polars:
     output: pyproject
     matrix:
@@ -724,6 +732,10 @@ files:
       polars_compat_version: ["1.35", "1.36", "1.37", "1.38", "1.39", "1.40", "1.41", "1.42"]
     includes:
       - test_cudf_polars_compat
+  test_cudf_polars_tpch:
+    output: none
+    includes:
+      - test_cudf_polars_tpch
 channels:
   - rapidsai-nightly
   - rapidsai
@@ -1321,6 +1333,15 @@ dependencies:
         packages:
           # Used by timeout_with_stack.py utility
           - gdb
+  test_cudf_polars_tpch:
+    common:
+      - output_types: [conda, requirements, pyproject]
+        packages:
+          - duckdb
+          - pyarrow
+          # added lower bound pinning to get pytest-subtests for free
+          - pytest>=9
+          - tpchgen-cli>=3.0.0
   test_python_narwhals:
     common:
       - output_types: [conda, requirements, constraints, pyproject]

python/cudf_polars/cudf_polars/streaming/benchmarks/asserts.py

@@ -324,6 +324,10 @@ def sort_for_comparison(df: pl.DataFrame) -> pl.DataFrame:
                         pl.col(col).lt(val - 2 * abs_tol)
                         | pl.col(col).gt(val + 2 * abs_tol)
                     )
+                elif val is None:
+                    filter_exprs.append(
+                        pl.col(col).is_not_null() if nulls_last else pl.lit(value=False)
+                    )
                 else:
                     if desc:
                         # then "before" means "greater than"

python/cudf_polars/cudf_polars/streaming/benchmarks/pdsds.py

@@ -31,18 +31,10 @@
     if e.name is not None and not e.name.startswith("cudf_polars"):
         raise
 
+
 if TYPE_CHECKING:
     from types import ModuleType
 
-# Without this setting, the first IO task to run
-# on each worker takes ~15 sec extra
-os.environ["KVIKIO_COMPAT_MODE"] = os.environ.get("KVIKIO_COMPAT_MODE", "on")
-os.environ["KVIKIO_NTHREADS"] = os.environ.get("KVIKIO_NTHREADS", "8")
-# TODO: consider raising the rapidsmpf built-in default from 1 to 8.
-os.environ["RAPIDSMPF_NUM_STREAMING_THREADS"] = os.environ.get(
-    "RAPIDSMPF_NUM_STREAMING_THREADS", "8"
-)
-
 
 def valid_query(name: str) -> bool:
     """Return True for valid query names eg. 'q9', 'q65', etc."""
@@ -75,12 +67,18 @@ class PDSDSQueries(metaclass=PDSDSQueriesMeta):
 
     q_impl: str
     name: str = "pdsds"
+    num_queries: int = 99
 
 
 class PDSDSPolarsQueries(PDSDSQueries):
     """Polars Queries."""
 
     q_impl = "polars_impl"
+    # Queries expected to fail on GPU due to known bugs. Keys are query numbers;
+    # values are reasons for the failures. These queries will be skipped in GPU runs.
+    EXPECTED_FAILURES_TPCDS: ClassVar[dict[int, str]] = {
+        5: "GPU execution failure (packed data cannot be empty): https://github.com/rapidsai/cudf/issues/22073",
+    }
     # See comments for EXPECTED_CASTS and EXPECTED_CASTS_DECIMAL
     # in cudf/python/cudf_polars/cudf_polars/streaming/benchmarks/pdsh.py
     # for more details.
@@ -111,18 +109,9 @@ class PDSDSPolarsQueries(PDSDSQueries):
             pl.col("total net profit").cast(pl.Decimal(18, 2)),
         ],
         19: [pl.col("ext_price").cast(pl.Decimal(18, 2))],
-        20: [
-            pl.col("itemrevenue").cast(pl.Decimal(18, 2)),
-            pl.col("revenueratio").cast(pl.Decimal(38, 2)),
-        ],
+        20: [pl.col("revenueratio").cast(pl.Decimal(38, 2))],
         24: [pl.col("paid").cast(pl.Decimal(18, 2))],
         30: [pl.col("ctr_total_return").cast(pl.Decimal(18, 2))],
-        31: [
-            pl.col("web_q1_q2_increase").cast(pl.Decimal(38, 2)),
-            pl.col("store_q1_q2_increase").cast(pl.Decimal(38, 2)),
-            pl.col("web_q2_q3_increase").cast(pl.Decimal(38, 2)),
-            pl.col("store_q2_q3_increase").cast(pl.Decimal(38, 2)),
-        ],
         32: [pl.col("excess discount amount").cast(pl.Decimal(18, 2))],
         33: [pl.col("total_sales").cast(pl.Decimal(18, 2))],
         42: [pl.col("sum(ss_ext_sales_price)").cast(pl.Decimal(18, 2))],
@@ -239,6 +228,11 @@ class PDSDSPolarsQueries(PDSDSQueries):
             pl.col("inv_before").cast(pl.Int32),
             pl.col("inv_after").cast(pl.Int32),
         ],
+        29: [
+            pl.col("store_sales_quantity").cast(pl.Int64),
+            pl.col("store_returns_quantity").cast(pl.Int64),
+            pl.col("catalog_sales_quantity").cast(pl.Int64),
+        ],
         34: [pl.col("cnt").cast(COUNT_DTYPE)],
         35: [
             pl.col("cnt1").cast(COUNT_DTYPE),
@@ -331,7 +325,16 @@ class PDSDSDuckDBQueries(PDSDSQueries):
 
 
 if __name__ == "__main__":
-    parser = build_parser(num_queries=99)
+    # Without this setting, the first IO task to run
+    # on each worker takes ~15 sec extra
+    os.environ["KVIKIO_COMPAT_MODE"] = os.environ.get("KVIKIO_COMPAT_MODE", "on")
+    os.environ["KVIKIO_NTHREADS"] = os.environ.get("KVIKIO_NTHREADS", "8")
+    # TODO: consider raising the rapidsmpf built-in default from 1 to 8.
+    os.environ["RAPIDSMPF_NUM_STREAMING_THREADS"] = os.environ.get(
+        "RAPIDSMPF_NUM_STREAMING_THREADS", "8"
+    )
+
+    parser = build_parser(num_queries=PDSDSQueries.num_queries)
     args = parse_args(parser=parser)
     if args.frontend not in _CPU_ENGINES:
         os.environ["POLARS_MAX_THREADS"] = os.environ.get("POLARS_MAX_THREADS", "1")

python/cudf_polars/cudf_polars/streaming/benchmarks/pdsds_queries/__init__.py

@@ -1,4 +1,27 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 """DuckDB and Polars queries."""
+
+from __future__ import annotations
+
+import polars as pl
+
+
+def sql_sum(expr: str | pl.Expr) -> pl.Expr:
+    """
+    Sum that returns NULL for all-null/empty groups, matching SQL SUM semantics.
+
+    Polars sum() returns 0 for all-null or empty groups; SQL returns NULL.
+    See https://github.com/rapidsai/cudf/issues/19560.
+
+    Parameters
+    ----------
+    expr
+        Column name or expression to sum. If a string, wraps in ``pl.col``.
+        Pass a conditional expression (e.g. ``pl.when(...).then(...).otherwise(None)``)
+        to implement SQL ``SUM(CASE WHEN ... END)`` without ``.filter()`` inside
+        a groupby, which is not supported on GPU.
+    """
+    e = pl.col(expr) if isinstance(expr, str) else expr
+    return pl.when(e.count() > 0).then(e.sum()).otherwise(None)

python/cudf_polars/cudf_polars/streaming/benchmarks/pdsds_queries/q1.py

@@ -10,6 +10,7 @@
 import polars as pl
 
 from cudf_polars.streaming.benchmarks.pdsds_parameters import load_parameters
+from cudf_polars.streaming.benchmarks.pdsds_queries import sql_sum
 from cudf_polars.streaming.benchmarks.utils import QueryResult, get_data
 
 if TYPE_CHECKING:
@@ -74,14 +75,7 @@ def polars_impl(run_config: RunConfig) -> QueryResult:
         )
         .filter(pl.col("d_year") == year)
         .group_by(["sr_customer_sk", "sr_store_sk"])
-        .agg(
-            # Polars sum() returns 0 for all-null groups; SQL returns NULL.
-            # See https://github.com/NVIDIA/cudf/issues/19560.
-            pl.when(pl.col("sr_return_amt").count() > 0)
-            .then(pl.col("sr_return_amt").sum())
-            .otherwise(None)
-            .alias("ctr_total_return")
-        )
+        .agg(sql_sum("sr_return_amt").alias("ctr_total_return"))
         .rename(
             {
                 "sr_customer_sk": "ctr_customer_sk",

python/cudf_polars/cudf_polars/streaming/benchmarks/pdsds_queries/q12.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 """Query 12."""
@@ -11,6 +11,7 @@
 import polars as pl
 
 from cudf_polars.streaming.benchmarks.pdsds_parameters import load_parameters
+from cudf_polars.streaming.benchmarks.pdsds_queries import sql_sum
 from cudf_polars.streaming.benchmarks.utils import QueryResult, get_data
 
 if TYPE_CHECKING:
@@ -95,23 +96,13 @@ def polars_impl(run_config: RunConfig) -> QueryResult:
             .group_by(
                 ["i_item_id", "i_item_desc", "i_category", "i_class", "i_current_price"]
             )
-            .agg(
-                [
-                    pl.when(pl.col("ws_ext_sales_price").count() > 0)
-                    .then(pl.col("ws_ext_sales_price").sum())
-                    .otherwise(None)
-                    .alias("itemrevenue")
-                ]
-            )
+            .agg([sql_sum("ws_ext_sales_price").alias("itemrevenue")])
             .with_columns(
                 [
                     (
                         pl.col("itemrevenue")
                         * 100
-                        / pl.when(pl.col("itemrevenue").count() > 0)
-                        .then(pl.col("itemrevenue").sum())
-                        .otherwise(None)
-                        .over("i_class")
+                        / sql_sum("itemrevenue").over("i_class")
                     ).alias("revenueratio")
                 ]
             )

python/cudf_polars/cudf_polars/streaming/benchmarks/pdsds_queries/q13.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 """Query 13."""
@@ -10,6 +10,7 @@
 import polars as pl
 
 from cudf_polars.streaming.benchmarks.pdsds_parameters import load_parameters
+from cudf_polars.streaming.benchmarks.pdsds_queries import sql_sum
 from cudf_polars.streaming.benchmarks.utils import QueryResult, get_data
 
 if TYPE_CHECKING:
@@ -158,9 +159,9 @@ def polars_impl(run_config: RunConfig) -> QueryResult:
                     pl.col("ss_ext_wholesale_cost")
                     .mean()
                     .alias("avg(ss_ext_wholesale_cost)"),
-                    pl.col("ss_ext_wholesale_cost")
-                    .sum()
-                    .alias("sum(ss_ext_wholesale_cost)"),
+                    sql_sum("ss_ext_wholesale_cost").alias(
+                        "sum(ss_ext_wholesale_cost)"
+                    ),
                 ]
             )
         ),