From 165ccc1a89eef905d88e20c2c5f5e6e85d10d2d0 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Tue, 9 Jun 2026 12:52:02 +0000 Subject: [PATCH 01/10] Drop dynamic predicates from logical plan --- .../cudf_polars/cudf_polars/dsl/translate.py | 47 +++++++++++++++---- .../cudf_polars/streaming/benchmarks/utils.py | 5 +- 2 files changed, 41 insertions(+), 11 deletions(-) diff --git a/python/cudf_polars/cudf_polars/dsl/translate.py b/python/cudf_polars/cudf_polars/dsl/translate.py index d1b86d40ee0a..3435cfd6a982 100644 --- a/python/cudf_polars/cudf_polars/dsl/translate.py +++ b/python/cudf_polars/cudf_polars/dsl/translate.py @@ -286,15 +286,43 @@ def __exit__(self, *args: Any) -> None: self.translator._expr_context = self._prev -def _is_dynamic_pred(visitor: Any, expr_ir: Any) -> bool: +def _drop_dyn_pred_hints( + translator: Translator, n: int, schema: Schema +) -> expr.Expr | None: + """Drop Polars' dynamic predicate hints.""" try: - visitor.view_expression(expr_ir.node) + node = translator.visitor.view_expression(n) except Exception as e: if str(e) == "dynamic_pred": - return True - raise # pragma: no cover - else: - return False + return None + raise + if isinstance(node, plrs._expr_nodes.BinaryExpr) and node.op in ( + plrs._expr_nodes.Operator.And, + plrs._expr_nodes.Operator.LogicalAnd, + ): + left = _drop_dyn_pred_hints(translator, node.left, schema) + right = _drop_dyn_pred_hints(translator, node.right, schema) + if left is None: + return right + if right is None: + return left + return expr.BinOp( + DataType(translator.visitor.get_dtype(n)), + expr.BinOp._MAPPING[node.op], + left, + right, + ) + return translator.translate_expr(n=n, schema=schema) + + +def translate_predicate( + translator: Translator, *, n: Any, schema: Schema +) -> expr.NamedExpr | None: + """Translate a filter predicate, dropping Polars' dynamic-predicate conjuncts.""" + mask = _drop_dyn_pred_hints(translator, n.node, schema) + if mask is None: + return None + return expr.NamedExpr(n.output_name, mask) class set_internal_name_gen(AbstractContextManager[None]): @@ -397,8 +425,7 @@ def _(node: plrs._ir_nodes.Scan, translator: Translator, schema: Schema) -> ir.I ( None if node.predicate is None - or _is_dynamic_pred(translator.visitor, node.predicate) - else translate_named_expr(translator, n=node.predicate, schema=schema) + else translate_predicate(translator, n=node.predicate, schema=schema) ), parquet_options, ) @@ -615,9 +642,9 @@ def _(node: plrs._ir_nodes.Slice, translator: Translator, schema: Schema) -> ir. def _(node: plrs._ir_nodes.Filter, translator: Translator, schema: Schema) -> ir.IR: with set_node(translator.visitor, node.input): inp = translator.translate_ir(n=None) - if _is_dynamic_pred(translator.visitor, node.predicate): + mask = translate_predicate(translator, n=node.predicate, schema=inp.schema) + if mask is None: return inp - mask = translate_named_expr(translator, n=node.predicate, schema=inp.schema) return ir.Filter(schema, mask, inp) diff --git a/python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py b/python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py index edf27ca272b5..2b331b06ba54 100644 --- a/python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py +++ b/python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py @@ -1164,9 +1164,9 @@ def run_polars_in_memory( **run_config.streaming_options.to_engine_options(), "parquet_options": parquet_options, } + engine_options.setdefault("raise_on_fail", True) engine = pl.GPUEngine( executor="in-memory", - raise_on_fail=True, **engine_options, ) records, plans, validation_failures, query_failures = _run_query_loop( @@ -1202,6 +1202,7 @@ def run_polars_spmd( **run_config.streaming_options.to_engine_options(), "parquet_options": parquet_options, } + engine_options.setdefault("raise_on_fail", True) with SPMDEngine( rapidsmpf_options=run_config.streaming_options.to_rapidsmpf_options(), executor_options=executor_options, @@ -1259,6 +1260,7 @@ def run_polars_ray( **run_config.streaming_options.to_engine_options(), "parquet_options": parquet_options, } + engine_options.setdefault("raise_on_fail", True) ray_init_options: dict[str, Any] = {} if run_config.connect is not None: ray_init_options["address"] = run_config.connect @@ -1308,6 +1310,7 @@ def run_polars_dask( **run_config.streaming_options.to_engine_options(), "parquet_options": parquet_options, } + engine_options.setdefault("raise_on_fail", True) dask_client = None if run_config.connect is not None: if Path(run_config.connect).is_file(): From 6f71b1b1912f900bc17c24e25de3881cd11b7ac6 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Tue, 9 Jun 2026 12:53:54 +0000 Subject: [PATCH 02/10] clean doc string --- python/cudf_polars/cudf_polars/dsl/translate.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/cudf_polars/cudf_polars/dsl/translate.py b/python/cudf_polars/cudf_polars/dsl/translate.py index 3435cfd6a982..3d99e3c3ef24 100644 --- a/python/cudf_polars/cudf_polars/dsl/translate.py +++ b/python/cudf_polars/cudf_polars/dsl/translate.py @@ -289,7 +289,6 @@ def __exit__(self, *args: Any) -> None: def _drop_dyn_pred_hints( translator: Translator, n: int, schema: Schema ) -> expr.Expr | None: - """Drop Polars' dynamic predicate hints.""" try: node = translator.visitor.view_expression(n) except Exception as e: @@ -318,7 +317,7 @@ def _drop_dyn_pred_hints( def translate_predicate( translator: Translator, *, n: Any, schema: Schema ) -> expr.NamedExpr | None: - """Translate a filter predicate, dropping Polars' dynamic-predicate conjuncts.""" + """Translate predicate, dropping Polars' dynamic predicate hints.""" mask = _drop_dyn_pred_hints(translator, n.node, schema) if mask is None: return None From a5b3383de3df0cf93f54c97c76b4d2ed7d20bc7e Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Tue, 9 Jun 2026 13:20:34 +0000 Subject: [PATCH 03/10] add a test --- python/cudf_polars/tests/test_filter.py | 36 +++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index 20dc656470f6..61700f7d2cda 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -5,7 +5,11 @@ import pytest import polars as pl +from polars import polars as plrs +from cudf_polars import Translator +from cudf_polars.dsl import expr, ir +from cudf_polars.dsl.traversal import traversal from cudf_polars.testing.asserts import assert_gpu_result_equal @@ -28,3 +32,35 @@ def test_filter(engine: pl.GPUEngine, expr, predicate_pushdown): "optimizations": pl.QueryOptFlags(predicate_pushdown=predicate_pushdown) }, ) + + +def test_filter_drops_dynamic_predicate_hint(): + class DynamicPredVisitor: + def __init__(self, inner, op): + self.inner = inner + self.op = op + + def __getattr__(self, name): + return getattr(self.inner, name) + + def view_expression(self, n): + node = self.inner.view_expression(n) + if isinstance(node, plrs._expr_nodes.BinaryExpr) and node.op == self.op: + raise Exception("dynamic_pred") + return node + + ldf = pl.LazyFrame( + {"a": [1, 2, 3, 4, 5], "b": [5, 4, 3, 2, 1], "c": [1, 1, 3, 3, 5]} + ) + query = ldf.filter( + (pl.col("b") < 5) & (pl.col("a") > 1) & (pl.col("c") == 3) & (pl.col("b") < 3) + ) + + visitor = DynamicPredVisitor(query._ldf.visit(), plrs._expr_nodes.Operator.Lt) + translator = Translator(visitor, pl.GPUEngine()) + node = translator.translate_ir() + + assert translator.errors == [] + assert isinstance(node, ir.Filter) + columns = {n.name for n in traversal([node.mask.value]) if isinstance(n, expr.Col)} + assert columns == {"a", "c"} From a04d464a9a5e2a7cf7dfda915b3409e2f31cddb0 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Tue, 9 Jun 2026 13:48:45 +0000 Subject: [PATCH 04/10] create a custom exception --- python/cudf_polars/tests/test_filter.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index 61700f7d2cda..f81322f878ec 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -35,6 +35,9 @@ def test_filter(engine: pl.GPUEngine, expr, predicate_pushdown): def test_filter_drops_dynamic_predicate_hint(): + class DynamicPredHint(Exception): + pass + class DynamicPredVisitor: def __init__(self, inner, op): self.inner = inner @@ -46,7 +49,7 @@ def __getattr__(self, name): def view_expression(self, n): node = self.inner.view_expression(n) if isinstance(node, plrs._expr_nodes.BinaryExpr) and node.op == self.op: - raise Exception("dynamic_pred") + raise DynamicPredHint("dynamic_pred") return node ldf = pl.LazyFrame( From a266a40ae37f700208cb18e5242e044841db3072 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Wed, 10 Jun 2026 14:41:47 +0000 Subject: [PATCH 05/10] mypy import check --- python/cudf_polars/tests/test_filter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index f81322f878ec..3138fcf60718 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -5,7 +5,7 @@ import pytest import polars as pl -from polars import polars as plrs +from polars import polars as plrs # type: ignore[attr-defined] from cudf_polars import Translator from cudf_polars.dsl import expr, ir From d68c3a8ae11a8a43e174f91a9d16eee642097fc2 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Thu, 11 Jun 2026 10:15:04 -0700 Subject: [PATCH 06/10] lint --- python/cudf_polars/tests/test_filter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index 3138fcf60718..ce154ba3bc07 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -5,7 +5,7 @@ import pytest import polars as pl -from polars import polars as plrs # type: ignore[attr-defined] +from polars import polars as plrs # type: ignore[attr-defined] from cudf_polars import Translator from cudf_polars.dsl import expr, ir From 0c9130e9e98c8a344bb4359400e48a699acf8bd2 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Fri, 12 Jun 2026 07:04:52 -0700 Subject: [PATCH 07/10] code coverge --- python/cudf_polars/cudf_polars/dsl/translate.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_polars/cudf_polars/dsl/translate.py b/python/cudf_polars/cudf_polars/dsl/translate.py index 3d99e3c3ef24..0e882ed0f448 100644 --- a/python/cudf_polars/cudf_polars/dsl/translate.py +++ b/python/cudf_polars/cudf_polars/dsl/translate.py @@ -294,7 +294,7 @@ def _drop_dyn_pred_hints( except Exception as e: if str(e) == "dynamic_pred": return None - raise + raise # pragma: no cover if isinstance(node, plrs._expr_nodes.BinaryExpr) and node.op in ( plrs._expr_nodes.Operator.And, plrs._expr_nodes.Operator.LogicalAnd, From 50286a05cb9b1997cd7cf867cc1fa5dd4ddf0eca Mon Sep 17 00:00:00 2001 From: Matthew Murray <41342305+Matt711@users.noreply.github.com> Date: Fri, 12 Jun 2026 21:58:46 -0400 Subject: [PATCH 08/10] extend fiter in test --- python/cudf_polars/tests/test_filter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index ce154ba3bc07..5aa91b290532 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -56,7 +56,7 @@ def view_expression(self, n): {"a": [1, 2, 3, 4, 5], "b": [5, 4, 3, 2, 1], "c": [1, 1, 3, 3, 5]} ) query = ldf.filter( - (pl.col("b") < 5) & (pl.col("a") > 1) & (pl.col("c") == 3) & (pl.col("b") < 3) + ((pl.col("b") < 5) & (pl.col("a") > 1)) & ((pl.col("c") == 3) & (pl.col("b") < 3)) ) visitor = DynamicPredVisitor(query._ldf.visit(), plrs._expr_nodes.Operator.Lt) From 4499867fae6b1e27f2d5caea280823b759f41fc7 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Mon, 15 Jun 2026 16:57:22 +0000 Subject: [PATCH 09/10] pre-commit check' --- python/cudf_polars/tests/test_filter.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index 5aa91b290532..6086a8f0cba7 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -56,7 +56,8 @@ def view_expression(self, n): {"a": [1, 2, 3, 4, 5], "b": [5, 4, 3, 2, 1], "c": [1, 1, 3, 3, 5]} ) query = ldf.filter( - ((pl.col("b") < 5) & (pl.col("a") > 1)) & ((pl.col("c") == 3) & (pl.col("b") < 3)) + ((pl.col("b") < 5) & (pl.col("a") > 1)) + & ((pl.col("c") == 3) & (pl.col("b") < 3)) ) visitor = DynamicPredVisitor(query._ldf.visit(), plrs._expr_nodes.Operator.Lt) From ff1e9b348043d2cb960d35169c7b7d142fd44c51 Mon Sep 17 00:00:00 2001 From: Matthew Murray Date: Mon, 15 Jun 2026 17:33:40 +0000 Subject: [PATCH 10/10] add a proper polars test --- python/cudf_polars/tests/test_filter.py | 42 +++++-------------------- 1 file changed, 7 insertions(+), 35 deletions(-) diff --git a/python/cudf_polars/tests/test_filter.py b/python/cudf_polars/tests/test_filter.py index 6086a8f0cba7..71781454e70d 100644 --- a/python/cudf_polars/tests/test_filter.py +++ b/python/cudf_polars/tests/test_filter.py @@ -5,11 +5,7 @@ import pytest import polars as pl -from polars import polars as plrs # type: ignore[attr-defined] -from cudf_polars import Translator -from cudf_polars.dsl import expr, ir -from cudf_polars.dsl.traversal import traversal from cudf_polars.testing.asserts import assert_gpu_result_equal @@ -34,37 +30,13 @@ def test_filter(engine: pl.GPUEngine, expr, predicate_pushdown): ) -def test_filter_drops_dynamic_predicate_hint(): - class DynamicPredHint(Exception): - pass - - class DynamicPredVisitor: - def __init__(self, inner, op): - self.inner = inner - self.op = op - - def __getattr__(self, name): - return getattr(self.inner, name) - - def view_expression(self, n): - node = self.inner.view_expression(n) - if isinstance(node, plrs._expr_nodes.BinaryExpr) and node.op == self.op: - raise DynamicPredHint("dynamic_pred") - return node - +def test_filter_drops_dynamic_predicate_hint(engine: pl.GPUEngine): ldf = pl.LazyFrame( {"a": [1, 2, 3, 4, 5], "b": [5, 4, 3, 2, 1], "c": [1, 1, 3, 3, 5]} ) - query = ldf.filter( - ((pl.col("b") < 5) & (pl.col("a") > 1)) - & ((pl.col("c") == 3) & (pl.col("b") < 3)) - ) - - visitor = DynamicPredVisitor(query._ldf.visit(), plrs._expr_nodes.Operator.Lt) - translator = Translator(visitor, pl.GPUEngine()) - node = translator.translate_ir() - - assert translator.errors == [] - assert isinstance(node, ir.Filter) - columns = {n.name for n in traversal([node.mask.value]) if isinstance(n, expr.Col)} - assert columns == {"a", "c"} + # sort("b").head(3) causes Polars to inject a dynamic predicate hint into + # the filter below: FILTER (a > 1) & (c == 3) & col("b").dynamic_predicate() + # This test ensures we drop these dynamic predicate hints from the filter + # before executing on the GPU. + q = ldf.filter((pl.col("a") > 1) & (pl.col("c") == 3)).sort("b").head(3) + assert_gpu_result_equal(q, engine=engine)