From 2039a231c640a503b9e8b6dd22dfcaf770624942 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Thu, 14 May 2026 16:12:52 +0200 Subject: [PATCH 01/10] --frontend=cpu|in-memory|spmd|ray|dask|duckdb --- .../benchmarks/utils_new_frontends.py | 152 +++++++++++++----- 1 file changed, 111 insertions(+), 41 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 378d04a260c2..fdcea0ba990c 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -23,7 +23,7 @@ from datetime import UTC, datetime from pathlib import Path from statistics import mean -from typing import TYPE_CHECKING, Any, Literal, assert_never +from typing import TYPE_CHECKING, Any, Literal import nvtx @@ -109,7 +109,7 @@ def get_validation_options(args: Any) -> dict[str, Any]: _HAS_STRUCTLOG = True -ExecutorType = Literal["in-memory", "streaming", "cpu"] +_STREAMING_FRONTENDS = frozenset({"spmd", "ray", "dask"}) @dataclasses.dataclass @@ -387,8 +387,7 @@ class RunConfig: qualification: bool = False # Execution mode - executor: ExecutorType # "in-memory" | "streaming" | "cpu" - frontend: str # "spmd" | "ray" | "duckdb" + frontend: Literal["cpu", "in-memory", "spmd", "ray", "dask", "duckdb"] connect: str | None = None num_gpus: int | None = None @@ -512,7 +511,7 @@ def from_args(cls, args: argparse.Namespace) -> RunConfig: if args.engine == "duckdb": engine_name = "duckdb" elif args.engine == "polars": - if args.executor == "cpu": + if args.frontend == "cpu": engine_name = "polars-cpu" else: engine_name = "cudf-polars" @@ -527,7 +526,6 @@ def from_args(cls, args: argparse.Namespace) -> RunConfig: scale_factor=scale_factor, suffix=args.suffix, qualification=args.qualification, - executor=args.executor, frontend=args.frontend, iterations=args.iterations, io_mode=args.io_mode, @@ -555,7 +553,6 @@ def serialize(self, engine: pl.GPUEngine | None) -> dict: "scale_factor": self.scale_factor, "suffix": self.suffix, "qualification": self.qualification, - "executor": self.executor, "frontend": self.frontend, "iterations": self.iterations, "io_mode": self.io_mode, @@ -595,9 +592,8 @@ def summarize(self) -> None: print(f"query: {query}") print(f"path: {self.dataset_path}") print(f"scale_factor: {self.scale_factor}") - print(f"executor: {self.executor}") print(f"frontend: {self.frontend}") - if self.executor == "streaming": + if self.frontend in _STREAMING_FRONTENDS: opts = self.streaming_options.to_executor_options() print(f"native_parquet: {self.native_parquet}") print(f"n_workers: {self.n_workers}") @@ -652,7 +648,7 @@ def print_query_plan( ) -> tuple[str | None, str | None]: """Print the query plan.""" logical_plan = plan = None - if run_config.executor == "cpu": + if run_config.frontend == "cpu": if args.explain_logical: logical_plan = q.explain() if args.explain: @@ -661,7 +657,7 @@ def print_query_plan( assert isinstance(engine, pl.GPUEngine) if args.explain_logical: logical_plan = explain_query(q, engine, physical=False) - if args.explain and run_config.executor == "streaming": + if args.explain and run_config.frontend in _STREAMING_FRONTENDS: plan = explain_query(q, engine) else: raise RuntimeError( @@ -714,7 +710,7 @@ def execute_query( domain="cudf_polars", color="green", ): - if run_config.executor == "cpu": + if run_config.frontend == "cpu": t0 = time.monotonic() result = q.collect(engine="streaming") t1 = time.monotonic() @@ -725,13 +721,13 @@ def execute_query( translator = Translator(q._ldf.visit(), engine) ir = translator.translate_ir() context = IRExecutionContext() - if run_config.executor == "in-memory": + if run_config.frontend == "in-memory": t0 = time.monotonic() result = ir.evaluate( cache={}, timer=None, context=context ).to_polars() t1 = time.monotonic() - elif run_config.executor == "streaming": + elif run_config.frontend in _STREAMING_FRONTENDS: t0 = time.monotonic() result = evaluate_streaming( ir, @@ -739,7 +735,9 @@ def execute_query( ) t1 = time.monotonic() else: - assert_never(run_config.executor) + raise ValueError( + f"--debug is not supported with --frontend {run_config.frontend}" + ) else: t0 = time.monotonic() result = q.collect(engine=engine) @@ -814,9 +812,15 @@ class QueryResult: sort_keys: list[tuple[pl.Expr, bool]] | None = None -def _collect_statistics(engine: StreamingEngine | None) -> dict[str, Any] | None: +def _collect_statistics(engine: pl.GPUEngine | None) -> dict[str, Any] | None: """Gather + clear per-rank rapidsmpf statistics into a merged dict.""" - return None if engine is None else engine.global_statistics(clear=True).to_dict() + if engine is None: + return None + from cudf_polars.experimental.rapidsmpf.frontend.core import StreamingEngine + + if not isinstance(engine, StreamingEngine): + return None + return engine.global_statistics(clear=True).to_dict() def run_polars_query_iteration( @@ -825,7 +829,7 @@ def run_polars_query_iteration( q: pl.LazyFrame, run_config: RunConfig, args: argparse.Namespace, - engine: StreamingEngine | None, + engine: pl.GPUEngine | None, expected: pl.DataFrame | None, query_result: Any, prepare_validation_result: Callable[[pl.DataFrame], pl.DataFrame] | None = None, @@ -882,7 +886,7 @@ def run_polars_query( benchmark: Any, run_config: RunConfig, args: argparse.Namespace, - engine: StreamingEngine | None, + engine: pl.GPUEngine | None, numeric_type: str, date_type: str, validation_files: dict[int, Path] | None, @@ -943,7 +947,11 @@ def run_polars_query( for i in range(args.iterations): if _HAS_STRUCTLOG and run_config.collect_traces: setup_logging(q_id, i) - if engine is not None: + from cudf_polars.experimental.rapidsmpf.frontend.core import ( + StreamingEngine, + ) + + if isinstance(engine, StreamingEngine): engine._run(setup_logging, q_id, i) try: @@ -999,7 +1007,7 @@ def _run_query_loop( benchmark: Any, args: argparse.Namespace, run_config: RunConfig, - engine: StreamingEngine | None, + engine: pl.GPUEngine | None, numeric_type: str, date_type: str, validation_files: dict[int, Path] | None, @@ -1064,7 +1072,7 @@ def _finalize_benchmark_run( """Summarize, serialize, and exit after a benchmark run.""" if args.summarize: run_config.summarize() - if args.validate and run_config.executor != "cpu": + if args.validate and run_config.frontend != "cpu": print("\nValidation Summary") print("==================") if validation_failures: @@ -1079,6 +1087,60 @@ def _finalize_benchmark_run( sys.exit(1 if (query_failures or validation_failures) else 0) +def run_polars_cpu( + benchmark: Any, + args: argparse.Namespace, + run_config: Any, + numeric_type: str, + date_type: str, + validation_files: dict[int, Path] | None, +) -> None: + """Run benchmark queries using the Polars CPU streaming engine.""" + records, plans, validation_failures, query_failures = _run_query_loop( + benchmark, + args, + run_config, + engine=None, + numeric_type=numeric_type, + date_type=date_type, + validation_files=validation_files, + ) + run_config = dataclasses.replace(run_config, records=dict(records), plans=plans) + _finalize_benchmark_run(args, run_config, validation_failures, query_failures) + + +def run_polars_in_memory( + benchmark: Any, + args: argparse.Namespace, + run_config: Any, + parquet_options: dict[str, Any], + numeric_type: str, + date_type: str, + validation_files: dict[int, Path] | None, +) -> None: + """Run benchmark queries using a single-process GPU in-memory engine.""" + engine_options = { + **run_config.streaming_options.to_engine_options(), + "parquet_options": parquet_options, + } + engine = pl.GPUEngine( + executor="in-memory", + raise_on_fail=True, + **engine_options, + ) + records, plans, validation_failures, query_failures = _run_query_loop( + benchmark, + args, + run_config, + engine=engine, + numeric_type=numeric_type, + date_type=date_type, + validation_files=validation_files, + ) + run_config = dataclasses.replace(run_config, records=dict(records), plans=plans) + _finalize_benchmark_run(args, run_config, validation_failures, query_failures) + + def run_polars_spmd( benchmark: Any, args: argparse.Namespace, @@ -1626,12 +1688,11 @@ def parse(query: str | int) -> list[int]: def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: - """Build the argument parser for PDS-H/PDS-DS benchmarks (new-frontend).""" + """Build the argument parser for PDS-H/PDS-DS benchmarks.""" from cudf_polars.experimental.rapidsmpf.frontend.options import StreamingOptions parser = argparse.ArgumentParser( prog="Cudf-Polars PDS-H/PDS-DS Benchmarks", - description="Experimental streaming-executor benchmarks (SPMD / Ray / DuckDB).", formatter_class=argparse.RawTextHelpFormatter, ) parser.add_argument( @@ -1677,29 +1738,19 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: File suffix for input table files. Default: .parquet"""), ) - parser.add_argument( - "-e", - "--executor", - default="streaming", - type=str, - choices=["in-memory", "streaming", "cpu"], - help=textwrap.dedent("""\ - Query executor backend: - - in-memory : Evaluate query in GPU memory - - streaming : Partitioned evaluation (default) - - cpu : Use Polars CPU engine"""), - ) parser.add_argument( "--frontend", required=True, type=str, - choices=["spmd", "ray", "dask", "duckdb"], + choices=["cpu", "in-memory", "spmd", "ray", "dask", "duckdb"], help=textwrap.dedent("""\ Execution frontend: - - spmd : SPMD execution via rrun launcher - - ray : Ray actor-based multi-GPU execution - - dask : Dask distributed multi-GPU execution - - duckdb : DuckDB CPU execution"""), + - cpu : Polars CPU streaming engine (no GPU) + - in-memory : Single-process GPU, in-memory evaluation + - spmd : SPMD execution via rrun launcher + - ray : Ray actor-based multi-GPU execution + - dask : Dask distributed multi-GPU execution + - duckdb : DuckDB CPU execution"""), ) parser.add_argument( "--connect", @@ -1961,6 +2012,25 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: ) numeric_type, date_type = check_input_data_type(run_config) match args.frontend: + case "cpu": + run_polars_cpu( + benchmark, + args, + run_config, + numeric_type, + date_type, + validation_files, + ) + case "in-memory": + run_polars_in_memory( + benchmark, + args, + run_config, + parquet_options, + numeric_type, + date_type, + validation_files, + ) case "spmd": run_polars_spmd( benchmark, From b99d3f54b8bce841166f8e454bfa88852af16d03 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Fri, 15 May 2026 09:32:21 +0200 Subject: [PATCH 02/10] sort the frontends --- .../benchmarks/utils_new_frontends.py | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index fdcea0ba990c..0b2d0a84facf 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -109,7 +109,7 @@ def get_validation_options(args: Any) -> dict[str, Any]: _HAS_STRUCTLOG = True -_STREAMING_FRONTENDS = frozenset({"spmd", "ray", "dask"}) +_STREAMING_FRONTENDS = frozenset({"dask", "ray", "spmd"}) @dataclasses.dataclass @@ -387,7 +387,7 @@ class RunConfig: qualification: bool = False # Execution mode - frontend: Literal["cpu", "in-memory", "spmd", "ray", "dask", "duckdb"] + frontend: Literal["cpu", "dask", "duckdb", "in-memory", "ray", "spmd"] connect: str | None = None num_gpus: int | None = None @@ -1742,15 +1742,15 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: "--frontend", required=True, type=str, - choices=["cpu", "in-memory", "spmd", "ray", "dask", "duckdb"], + choices=["cpu", "dask", "duckdb", "in-memory", "ray", "spmd"], help=textwrap.dedent("""\ Execution frontend: - cpu : Polars CPU streaming engine (no GPU) + - dask : Dask distributed multi-GPU execution + - duckdb : DuckDB CPU execution - in-memory : Single-process GPU, in-memory evaluation - - spmd : SPMD execution via rrun launcher - ray : Ray actor-based multi-GPU execution - - dask : Dask distributed multi-GPU execution - - duckdb : DuckDB CPU execution"""), + - spmd : SPMD execution via rrun launcher"""), ) parser.add_argument( "--connect", @@ -1994,7 +1994,7 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: if run_config.num_gpus is not None: if run_config.connect is not None: raise ValueError("--num-gpus cannot be used with --connect.") - if run_config.frontend not in ("ray", "dask"): + if run_config.frontend not in ("dask", "ray"): raise ValueError( "--num-gpus is only supported with --frontend ray or dask." ) @@ -2021,8 +2021,8 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "in-memory": - run_polars_in_memory( + case "dask": + run_polars_dask( benchmark, args, run_config, @@ -2031,8 +2031,10 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "spmd": - run_polars_spmd( + case "duckdb": + run_duckdb(benchmark, args) + case "in-memory": + run_polars_in_memory( benchmark, args, run_config, @@ -2051,8 +2053,8 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "dask": - run_polars_dask( + case "spmd": + run_polars_spmd( benchmark, args, run_config, @@ -2061,7 +2063,5 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "duckdb": - run_duckdb(benchmark, args) case _: raise ValueError(f"Unknown --frontend: {args.frontend!r}") From 6b4e968884a21aec343d01face13edad0ffaf06d Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Sat, 16 May 2026 09:29:08 +0200 Subject: [PATCH 03/10] unify the cpu frontends duckdb-cpu and polars-cpu --- .../experimental/benchmarks/pdsds.py | 15 +-- .../experimental/benchmarks/pdsh.py | 15 +-- .../benchmarks/utils_new_frontends.py | 100 +++++++++--------- 3 files changed, 52 insertions(+), 78 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsds.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsds.py index c73574b397ed..9d6d7ee5f7cd 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsds.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsds.py @@ -24,7 +24,6 @@ COUNT_DTYPE, build_parser, parse_args, - run_duckdb, run_polars, ) except ImportError as e: @@ -328,17 +327,5 @@ class PDSDSDuckDBQueries(PDSDSQueries): if __name__ == "__main__": parser = build_parser(num_queries=99) - parser.add_argument( - "--engine", - choices=["polars", "duckdb"], - default="polars", - help="Which engine to use for executing the benchmarks or to validate results.", - ) args = parse_args(parser=parser) - - if args.engine == "polars": - run_polars(PDSDSPolarsQueries, args) - elif args.engine == "duckdb": - run_duckdb(PDSDSDuckDBQueries, args) - else: - raise ValueError(f"Invalid engine: {args.engine}") + run_polars(PDSDSPolarsQueries, args) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsh.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsh.py index 9f15f137bb1d..59a71fe8d139 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsh.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/pdsh.py @@ -27,7 +27,6 @@ build_parser, get_data, parse_args, - run_duckdb, run_polars, ) except ImportError as e: @@ -1795,17 +1794,5 @@ def q22(run_config: RunConfig) -> str: if __name__ == "__main__": parser = build_parser(num_queries=22) - parser.add_argument( - "--engine", - choices=["polars", "duckdb"], - default="polars", - help="Which engine to use for executing the benchmarks or to validate results.", - ) args = parse_args(parser=parser) - - if args.engine == "polars": - run_polars(PDSHQueries, args) - elif args.engine == "duckdb": - run_duckdb(PDSHDuckDBQueries, args) - else: - raise ValueError(f"Invalid engine: {args.engine}") + run_polars(PDSHQueries, args) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 0b2d0a84facf..38c1a7bb095b 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -180,7 +180,7 @@ class ValidationMethod: things like the tolerance for floating point comparisons. """ - expected_source: Literal["polars-cpu", "duckdb"] + expected_source: Literal["polars-cpu", "duckdb-cpu"] comparison_method: Literal["polars"] comparison_options: dict[str, Any] @@ -377,7 +377,7 @@ def _infer_scale_factor(name: str, path: str | Path, suffix: str) -> int | float class RunConfig: """Benchmark run configuration for SPMD / Ray / DuckDB frontends.""" - engine_name: Literal["polars-cpu", "cudf-polars", "duckdb"] + engine_name: Literal["polars-cpu", "cudf-polars", "duckdb-cpu"] # Query selection & dataset queries: list[int] query_set: str @@ -387,7 +387,7 @@ class RunConfig: qualification: bool = False # Execution mode - frontend: Literal["cpu", "dask", "duckdb", "in-memory", "ray", "spmd"] + frontend: Literal["dask", "duckdb-cpu", "in-memory", "polars-cpu", "ray", "spmd"] connect: str | None = None num_gpus: int | None = None @@ -494,29 +494,28 @@ def from_args(cls, args: argparse.Namespace) -> RunConfig: if args.validate_directory: validation_method = ValidationMethod( - expected_source="duckdb", + expected_source="duckdb-cpu", comparison_method="polars", comparison_options=get_validation_options(args), ) elif args.validate: validation_method = ValidationMethod( - expected_source="polars-cpu" if args.baseline == "cpu" else "duckdb", + expected_source="polars-cpu" + if args.baseline == "polars-cpu" + else "duckdb-cpu", comparison_method="polars", comparison_options=get_validation_options(args), ) else: validation_method = None - engine_name: Literal["polars-cpu", "cudf-polars", "duckdb"] - if args.engine == "duckdb": - engine_name = "duckdb" - elif args.engine == "polars": - if args.frontend == "cpu": - engine_name = "polars-cpu" - else: - engine_name = "cudf-polars" + engine_name: Literal["polars-cpu", "cudf-polars", "duckdb-cpu"] + if args.frontend == "duckdb-cpu": + engine_name = "duckdb-cpu" + elif args.frontend == "polars-cpu": + engine_name = "polars-cpu" else: - raise ValueError(f"Invalid engine: {args.engine}") + engine_name = "cudf-polars" return cls( engine_name=engine_name, @@ -648,7 +647,7 @@ def print_query_plan( ) -> tuple[str | None, str | None]: """Print the query plan.""" logical_plan = plan = None - if run_config.frontend == "cpu": + if run_config.frontend == "polars-cpu": if args.explain_logical: logical_plan = q.explain() if args.explain: @@ -710,7 +709,7 @@ def execute_query( domain="cudf_polars", color="green", ): - if run_config.frontend == "cpu": + if run_config.frontend == "polars-cpu": t0 = time.monotonic() result = q.collect(engine="streaming") t1 = time.monotonic() @@ -911,21 +910,22 @@ def run_polars_query( expected: pl.DataFrame | None = None if args.validate: - if args.baseline == "cpu": - expected = q.collect() - elif args.baseline == "duckdb": - duckdb_queries_cls = benchmark().duckdb_queries - get_ddb = getattr(duckdb_queries_cls, f"q{q_id}") - base_sql = get_ddb(run_config) - expected = execute_duckdb_query( - base_sql, - run_config.dataset_path, - query_set=duckdb_queries_cls.name, - suffix=run_config.suffix, - run_config=run_config, - ).with_columns(*casts) - else: - raise ValueError(f"Invalid baseline: {args.baseline}") + match args.baseline: + case "polars-cpu": + expected = q.collect() + case "duckdb-cpu": + duckdb_queries_cls = benchmark().duckdb_queries + get_ddb = getattr(duckdb_queries_cls, f"q{q_id}") + base_sql = get_ddb(run_config) + expected = execute_duckdb_query( + base_sql, + run_config.dataset_path, + query_set=duckdb_queries_cls.name, + suffix=run_config.suffix, + run_config=run_config, + ).with_columns(*casts) + case _: + raise ValueError(f"Invalid baseline: {args.baseline}") elif validation_files is not None: expected = pl.read_parquet(validation_files[q_id]).with_columns(*casts) else: @@ -1072,7 +1072,7 @@ def _finalize_benchmark_run( """Summarize, serialize, and exit after a benchmark run.""" if args.summarize: run_config.summarize() - if args.validate and run_config.frontend != "cpu": + if args.validate and "cpu" not in run_config.frontend: print("\nValidation Summary") print("==================") if validation_failures: @@ -1742,15 +1742,15 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: "--frontend", required=True, type=str, - choices=["cpu", "dask", "duckdb", "in-memory", "ray", "spmd"], + choices=["dask", "duckdb-cpu", "in-memory", "polars-cpu", "ray", "spmd"], help=textwrap.dedent("""\ Execution frontend: - - cpu : Polars CPU streaming engine (no GPU) - - dask : Dask distributed multi-GPU execution - - duckdb : DuckDB CPU execution - - in-memory : Single-process GPU, in-memory evaluation - - ray : Ray actor-based multi-GPU execution - - spmd : SPMD execution via rrun launcher"""), + - dask : Dask distributed multi-GPU execution + - duckdb-cpu : DuckDB CPU execution + - in-memory : Single-process GPU, in-memory evaluation + - polars-cpu : Polars CPU streaming engine (no GPU) + - ray : Ray actor-based multi-GPU execution + - spmd : SPMD execution via rrun launcher"""), ) parser.add_argument( "--connect", @@ -1864,8 +1864,8 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: ) parser.add_argument( "--baseline", - choices=["duckdb", "cpu"], - default="duckdb", + choices=["duckdb-cpu", "polars-cpu"], + default="duckdb-cpu", help="Which engine to use as the baseline for validation.", ) parser.add_argument( @@ -2012,17 +2012,20 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: ) numeric_type, date_type = check_input_data_type(run_config) match args.frontend: - case "cpu": - run_polars_cpu( + case "dask": + run_polars_dask( benchmark, args, run_config, + parquet_options, numeric_type, date_type, validation_files, ) - case "dask": - run_polars_dask( + case "duckdb-cpu": + run_duckdb(benchmark().duckdb_queries, args) + case "in-memory": + run_polars_in_memory( benchmark, args, run_config, @@ -2031,14 +2034,11 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "duckdb": - run_duckdb(benchmark, args) - case "in-memory": - run_polars_in_memory( + case "polars-cpu": + run_polars_cpu( benchmark, args, run_config, - parquet_options, numeric_type, date_type, validation_files, From 721084fe8a99de87047dfc26853ac52486d1064d Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Sat, 16 May 2026 09:41:31 +0200 Subject: [PATCH 04/10] fix --collect-traces --- .../benchmarks/utils_new_frontends.py | 47 +++++++++++++++++-- 1 file changed, 42 insertions(+), 5 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 38c1a7bb095b..d88b0cf4c52b 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -1138,6 +1138,7 @@ def run_polars_in_memory( validation_files=validation_files, ) run_config = dataclasses.replace(run_config, records=dict(records), plans=plans) + run_config = _consolidate_logs(run_config, engine=None) _finalize_benchmark_run(args, run_config, validation_failures, query_failures) @@ -1381,9 +1382,32 @@ def inject( def _consolidate_logs( - run_config: RunConfig, engine: StreamingEngine, *, gather_client_logs: bool = True + run_config: RunConfig, + engine: StreamingEngine | None, + *, + gather_client_logs: bool = True, ) -> RunConfig: - """Merge structlog traces from the local process and Dask workers into run_config.""" + """ + Gather structlog traces and attach them to ``run_config.records``. + + Parameters + ---------- + run_config + The benchmark run config to augment. + engine + The streaming engine to fan out the gather across (dask / ray / spmd). + Pass ``None`` for single-process frontends (e.g. in-memory), only the + local-process buffer is collected. + gather_client_logs + When ``engine`` is not ``None``, also include the client-side + local-process buffer. Set to ``False`` for SPMD, where rank-0 is + itself a worker (so the worker fan-out already covered it). Ignored + when ``engine`` is ``None``. + + Returns + ------- + The augmented ``run_config``. + """ if not (_HAS_STRUCTLOG and run_config.collect_traces): return run_config @@ -1391,9 +1415,12 @@ def gather_logs() -> str: logger = logging.getLogger() return logger.handlers[0].stream.getvalue() # type: ignore[attr-defined] - all_logs = "\n".join(engine._run(gather_logs)) - if gather_client_logs: - all_logs += "\n" + gather_logs() + parts: list[str] = [] + if engine is not None: + parts.append("\n".join(engine._run(gather_logs))) + if engine is None or gather_client_logs: + parts.append(gather_logs()) + all_logs = "\n".join(parts) parsed_logs = [json.loads(log) for log in all_logs.splitlines() if log] # Some other log records can end up in here. Filter those out. @@ -1991,6 +2018,16 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: if run_config.connect is not None and run_config.frontend == "spmd": raise ValueError("--connect is not supported with --frontend spmd.") + if run_config.collect_traces and run_config.frontend in ( + "polars-cpu", + "duckdb-cpu", + ): + raise ValueError( + f"--collect-traces is not supported with --frontend {run_config.frontend}; " + "cudf-polars tracing only applies to GPU frontends " + "(in-memory, dask, ray, spmd)." + ) + if run_config.num_gpus is not None: if run_config.connect is not None: raise ValueError("--num-gpus cannot be used with --connect.") From e18f6036e2ead356644dec5b88fd1d83808f5b24 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Sat, 16 May 2026 10:00:11 +0200 Subject: [PATCH 05/10] coderabbit suggestions --- .../benchmarks/utils_new_frontends.py | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index d88b0cf4c52b..306142628d54 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -110,6 +110,7 @@ def get_validation_options(args: Any) -> dict[str, Any]: _STREAMING_FRONTENDS = frozenset({"dask", "ray", "spmd"}) +_CPU_ENGINES = frozenset({"polars-cpu", "duckdb-cpu"}) @dataclasses.dataclass @@ -1072,7 +1073,7 @@ def _finalize_benchmark_run( """Summarize, serialize, and exit after a benchmark run.""" if args.summarize: run_config.summarize() - if args.validate and "cpu" not in run_config.frontend: + if args.validate and run_config.frontend not in _CPU_ENGINES: print("\nValidation Summary") print("==================") if validation_failures: @@ -2015,19 +2016,24 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: vars(args).update({"query_set": benchmark.name}) run_config = RunConfig.from_args(args) - if run_config.connect is not None and run_config.frontend == "spmd": - raise ValueError("--connect is not supported with --frontend spmd.") + if run_config.connect is not None and run_config.frontend not in ("dask", "ray"): + raise ValueError("--connect is only supported with --frontend ray or dask.") - if run_config.collect_traces and run_config.frontend in ( - "polars-cpu", - "duckdb-cpu", - ): + if run_config.collect_traces and run_config.frontend in _CPU_ENGINES: raise ValueError( f"--collect-traces is not supported with --frontend {run_config.frontend}; " "cudf-polars tracing only applies to GPU frontends " "(in-memory, dask, ray, spmd)." ) + if run_config.validation_method is not None and run_config.frontend in _CPU_ENGINES: + raise ValueError( + f"--validate/--validate-directory is not supported with --frontend " + f"{run_config.frontend}; validation compares a candidate engine against " + "a CPU baseline, so it only applies to GPU frontends " + "(in-memory, dask, ray, spmd)." + ) + if run_config.num_gpus is not None: if run_config.connect is not None: raise ValueError("--num-gpus cannot be used with --connect.") From 6032420c93c3affa508e8acb8a622aeb58659e11 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Sat, 16 May 2026 10:11:22 +0200 Subject: [PATCH 06/10] docs --- .../cudf_polars/experimental/benchmarks/utils_new_frontends.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 306142628d54..1df5a3633aa5 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -169,7 +169,7 @@ class ValidationMethod: A name indicating the source of the expected results. - 'polars-cpu': Run polars against the same data - - 'duckdb': Compare against pre-computed DuckDB results + - 'duckdb-cpu': Compare against pre-computed DuckDB results comparison_method How the comparison was performed. Currently, only From c6c1cacbb3adb604f7a8511c92c976ee9eb49939 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Sat, 16 May 2026 10:32:09 +0200 Subject: [PATCH 07/10] coderabbitai --- .../benchmarks/utils_new_frontends.py | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 1df5a3633aa5..776434a8a634 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -1441,18 +1441,18 @@ def sort_key(x: dict) -> tuple[int, int]: ) for query_id, run_logs_group in grouped: - run_logs = list(run_logs_group) - by_iteration = [ - list(x) - for _, x in itertools.groupby(run_logs, key=lambda x: x["iteration"]) - ] + traces_by_iteration: dict[int, list[dict[str, Any]]] = { + iteration: list(group) + for iteration, group in itertools.groupby( + run_logs_group, key=lambda x: x["iteration"] + ) + } run_records = run_config.records[query_id] - assert len(by_iteration) == len(run_records) # same number of iterations - all_traces = [list(iteration) for iteration in by_iteration] new_records: list[SuccessRecord | FailedRecord] = [] - for rec, traces in zip(run_records, all_traces, strict=True): - if rec.status == "success": + for rec in run_records: + traces = traces_by_iteration.get(rec.iteration) + if rec.status == "success" and traces is not None: new_records.append(dataclasses.replace(rec, traces=traces)) else: new_records.append(rec) @@ -2034,6 +2034,12 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: "(in-memory, dask, ray, spmd)." ) + if args.debug and run_config.frontend in _CPU_ENGINES: + raise ValueError( + f"--debug is not supported with --frontend {run_config.frontend}; " + "debug mode only applies to GPU frontends (in-memory, dask, ray, spmd)." + ) + if run_config.num_gpus is not None: if run_config.connect is not None: raise ValueError("--num-gpus cannot be used with --connect.") From 8f10038ad51c51c273c43f3c13ee7518465aa51e Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Mon, 18 May 2026 14:15:38 +0200 Subject: [PATCH 08/10] renamed duckdb-cpu back to duckdb --- .../benchmarks/utils_new_frontends.py | 32 +++++++++---------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 776434a8a634..de1019021057 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -110,7 +110,7 @@ def get_validation_options(args: Any) -> dict[str, Any]: _STREAMING_FRONTENDS = frozenset({"dask", "ray", "spmd"}) -_CPU_ENGINES = frozenset({"polars-cpu", "duckdb-cpu"}) +_CPU_ENGINES = frozenset({"polars-cpu", "duckdb"}) @dataclasses.dataclass @@ -169,7 +169,7 @@ class ValidationMethod: A name indicating the source of the expected results. - 'polars-cpu': Run polars against the same data - - 'duckdb-cpu': Compare against pre-computed DuckDB results + - 'duckdb': Compare against pre-computed DuckDB results comparison_method How the comparison was performed. Currently, only @@ -181,7 +181,7 @@ class ValidationMethod: things like the tolerance for floating point comparisons. """ - expected_source: Literal["polars-cpu", "duckdb-cpu"] + expected_source: Literal["polars-cpu", "duckdb"] comparison_method: Literal["polars"] comparison_options: dict[str, Any] @@ -378,7 +378,7 @@ def _infer_scale_factor(name: str, path: str | Path, suffix: str) -> int | float class RunConfig: """Benchmark run configuration for SPMD / Ray / DuckDB frontends.""" - engine_name: Literal["polars-cpu", "cudf-polars", "duckdb-cpu"] + engine_name: Literal["polars-cpu", "cudf-polars", "duckdb"] # Query selection & dataset queries: list[int] query_set: str @@ -388,7 +388,7 @@ class RunConfig: qualification: bool = False # Execution mode - frontend: Literal["dask", "duckdb-cpu", "in-memory", "polars-cpu", "ray", "spmd"] + frontend: Literal["dask", "duckdb", "in-memory", "polars-cpu", "ray", "spmd"] connect: str | None = None num_gpus: int | None = None @@ -495,7 +495,7 @@ def from_args(cls, args: argparse.Namespace) -> RunConfig: if args.validate_directory: validation_method = ValidationMethod( - expected_source="duckdb-cpu", + expected_source="duckdb", comparison_method="polars", comparison_options=get_validation_options(args), ) @@ -503,16 +503,16 @@ def from_args(cls, args: argparse.Namespace) -> RunConfig: validation_method = ValidationMethod( expected_source="polars-cpu" if args.baseline == "polars-cpu" - else "duckdb-cpu", + else "duckdb", comparison_method="polars", comparison_options=get_validation_options(args), ) else: validation_method = None - engine_name: Literal["polars-cpu", "cudf-polars", "duckdb-cpu"] - if args.frontend == "duckdb-cpu": - engine_name = "duckdb-cpu" + engine_name: Literal["polars-cpu", "cudf-polars", "duckdb"] + if args.frontend == "duckdb": + engine_name = "duckdb" elif args.frontend == "polars-cpu": engine_name = "polars-cpu" else: @@ -914,7 +914,7 @@ def run_polars_query( match args.baseline: case "polars-cpu": expected = q.collect() - case "duckdb-cpu": + case "duckdb": duckdb_queries_cls = benchmark().duckdb_queries get_ddb = getattr(duckdb_queries_cls, f"q{q_id}") base_sql = get_ddb(run_config) @@ -1770,11 +1770,11 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: "--frontend", required=True, type=str, - choices=["dask", "duckdb-cpu", "in-memory", "polars-cpu", "ray", "spmd"], + choices=["dask", "duckdb", "in-memory", "polars-cpu", "ray", "spmd"], help=textwrap.dedent("""\ Execution frontend: - dask : Dask distributed multi-GPU execution - - duckdb-cpu : DuckDB CPU execution + - duckdb : DuckDB CPU execution - in-memory : Single-process GPU, in-memory evaluation - polars-cpu : Polars CPU streaming engine (no GPU) - ray : Ray actor-based multi-GPU execution @@ -1892,8 +1892,8 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: ) parser.add_argument( "--baseline", - choices=["duckdb-cpu", "polars-cpu"], - default="duckdb-cpu", + choices=["duckdb", "polars-cpu"], + default="duckdb", help="Which engine to use as the baseline for validation.", ) parser.add_argument( @@ -2071,7 +2071,7 @@ def run_polars(benchmark: Any, args: argparse.Namespace) -> None: date_type, validation_files, ) - case "duckdb-cpu": + case "duckdb": run_duckdb(benchmark().duckdb_queries, args) case "in-memory": run_polars_in_memory( From 1d4901b2ddfac841fb8c2e4804598b089790e933 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Mon, 18 May 2026 16:07:03 +0200 Subject: [PATCH 09/10] update --connect doc --- .../experimental/benchmarks/utils_new_frontends.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index de1019021057..3252fb2d4cef 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -1787,10 +1787,9 @@ def build_parser(num_queries: int = 22) -> argparse.ArgumentParser: type=str, help=textwrap.dedent("""\ Connect to an existing cluster instead of creating a local one. - For --frontend dask: a TCP address (e.g. tcp://host:8786) or a - scheduler file path. For --frontend ray: a Ray address - (e.g. ray://host:10001 or "auto"). - Not supported with --frontend spmd."""), + Only supported with --frontend dask or ray: + - dask : a TCP address (e.g. tcp://host:8786) or a scheduler file path + - ray : a Ray address (e.g. ray://host:10001 or "auto")"""), ) parser.add_argument( "--num-gpus", From 783a125589e058b30dd2dfd9bdbd901b39636372 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Mon, 18 May 2026 16:22:45 +0200 Subject: [PATCH 10/10] coderabbitai --- .../experimental/benchmarks/utils_new_frontends.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py index 3252fb2d4cef..77322e6bc7db 100644 --- a/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py +++ b/python/cudf_polars/cudf_polars/experimental/benchmarks/utils_new_frontends.py @@ -68,6 +68,7 @@ ) from cudf_polars.experimental.explain import explain_query from cudf_polars.experimental.parallel import evaluate_streaming + from cudf_polars.experimental.rapidsmpf.frontend.core import StreamingEngine from cudf_polars.utils.config import ConfigOptions CUDF_POLARS_AVAILABLE = True @@ -78,7 +79,6 @@ from collections.abc import Callable from cudf_polars.experimental.explain import SerializablePlan - from cudf_polars.experimental.rapidsmpf.frontend.core import StreamingEngine from cudf_polars.experimental.rapidsmpf.frontend.options import StreamingOptions POLARS_VALIDATION_OPTIONS = { "check_row_order": True, @@ -816,8 +816,6 @@ def _collect_statistics(engine: pl.GPUEngine | None) -> dict[str, Any] | None: """Gather + clear per-rank rapidsmpf statistics into a merged dict.""" if engine is None: return None - from cudf_polars.experimental.rapidsmpf.frontend.core import StreamingEngine - if not isinstance(engine, StreamingEngine): return None return engine.global_statistics(clear=True).to_dict() @@ -948,10 +946,6 @@ def run_polars_query( for i in range(args.iterations): if _HAS_STRUCTLOG and run_config.collect_traces: setup_logging(q_id, i) - from cudf_polars.experimental.rapidsmpf.frontend.core import ( - StreamingEngine, - ) - if isinstance(engine, StreamingEngine): engine._run(setup_logging, q_id, i) @@ -1073,7 +1067,10 @@ def _finalize_benchmark_run( """Summarize, serialize, and exit after a benchmark run.""" if args.summarize: run_config.summarize() - if args.validate and run_config.frontend not in _CPU_ENGINES: + if ( + run_config.validation_method is not None + and run_config.frontend not in _CPU_ENGINES + ): print("\nValidation Summary") print("==================") if validation_failures: