Bump polars upper bound to <1.40

conda/environments/all_cuda-129_arch-aarch64.yaml

@@ -67,7 +67,7 @@ dependencies:
 - packaging
 - pandas>=2.0,<2.4.0
 - pandoc
-- polars>=1.35,<1.39
+- polars>=1.35,<1.40
 - pre-commit
 - pyarrow>=19.0.0,<24
 - pydata-sphinx-theme>=0.15.4

conda/environments/all_cuda-129_arch-x86_64.yaml

@@ -67,7 +67,7 @@ dependencies:
 - packaging
 - pandas>=2.0,<2.4.0
 - pandoc
-- polars>=1.35,<1.39
+- polars>=1.35,<1.40
 - pre-commit
 - pyarrow>=19.0.0,<24
 - pydata-sphinx-theme>=0.15.4

conda/environments/all_cuda-132_arch-aarch64.yaml

@@ -67,7 +67,7 @@ dependencies:
 - packaging
 - pandas>=2.0,<2.4.0
 - pandoc
-- polars>=1.35,<1.39
+- polars>=1.35,<1.40
 - pre-commit
 - pyarrow>=19.0.0,<24
 - pydata-sphinx-theme>=0.15.4

conda/environments/all_cuda-132_arch-x86_64.yaml

@@ -67,7 +67,7 @@ dependencies:
 - packaging
 - pandas>=2.0,<2.4.0
 - pandoc
-- polars>=1.35,<1.39
+- polars>=1.35,<1.40
 - pre-commit
 - pyarrow>=19.0.0,<24
 - pydata-sphinx-theme>=0.15.4

conda/recipes/cudf-polars/recipe.yaml

@@ -37,7 +37,7 @@ requirements:
     - python
     - pylibcudf =${{ version }}
     - rapidsmpf =${{ minor_version }}
-    - polars>=1.35,<1.39
+    - polars>=1.35,<1.40
     - packaging
     - ${{ pin_compatible("cuda-version", upper_bound="x", lower_bound="x") }}
     - if: cuda_major == "12"

dependencies.yaml

@@ -838,7 +838,7 @@ dependencies:
           # 'nvidia-ml-py' provides the 'pynvml' module
           - &nvidia_ml_py nvidia-ml-py>=12
           - packaging
-          - polars>=1.35,<1.39
+          - polars>=1.35,<1.40
           - typing_extensions>=4.0.0
   run_cudf_polars_dask:
     common:

python/cudf_polars/cudf_polars/dsl/expressions/rolling.py

@@ -22,7 +22,7 @@
     offsets_to_windows,
     range_window_bounds,
 )
-from cudf_polars.utils.versions import POLARS_VERSION_LT_136
+from cudf_polars.utils.versions import POLARS_VERSION_LT_136, POLARS_VERSION_LT_139
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -79,14 +79,14 @@ def to_request(
     elif isinstance(value, expr.Agg):
         child = value.children[0]
         col = child.evaluate(df, context=ExecutionContext.ROLLING)
-        if POLARS_VERSION_LT_136 and value.name == "var":  # pragma: no cover
+        if (POLARS_VERSION_LT_136 or not POLARS_VERSION_LT_139) and value.name == "var":
             # Polars variance produces null if nvalues <= ddof
             # libcudf produces NaN. However, we can get the polars
             # behaviour by setting the minimum window size to ddof +
             # 1.
-            #
-            # We still need this check, polars is not hitting it because
-            # of https://github.com/pola-rs/polars/pull/25117
+            # In polars 1.36-1.38 this code path is not hit because
+            # rolling goes through the IR path (not RollingWindow).
+            # See https://github.com/pola-rs/polars/pull/25117
             min_periods = value.options + 1
     else:
         col = value.evaluate(
@@ -146,12 +146,12 @@ def __init__(
             raise NotImplementedError(
                 "Incorrect handling of empty groups for list collection"
             )
-        if POLARS_VERSION_LT_136 and not plc.rolling.is_valid_rolling_aggregation(
+        if (
+            POLARS_VERSION_LT_136 or not POLARS_VERSION_LT_139
+        ) and not plc.rolling.is_valid_rolling_aggregation(
             agg.dtype.plc_type, agg.agg_request
         ):
-            raise NotImplementedError(
-                f"Unsupported rolling aggregation {agg}"
-            )  # pragma: no cover; polars may raise ahead of time
+            raise NotImplementedError(f"Unsupported rolling aggregation {agg}")
 
     def do_evaluate(  # noqa: D102
         self, df: DataFrame, *, context: ExecutionContext = ExecutionContext.FRAME

python/cudf_polars/cudf_polars/dsl/expressions/unary.py

@@ -240,15 +240,14 @@ def do_evaluate(
             if maintain_order:
                 column = column.sorted_like(values)
             return column
-        elif self.name == "set_sorted":  # pragma: no cover
-            # TODO: LazyFrame.set_sorted is proper IR concept (ie. FunctionIR::Hint)
-            # and is is currently not implemented. We should reimplement it as a MapFunction.
+        elif self.name == "set_sorted":
             (column,) = (child.evaluate(df, context=context) for child in self.children)
-            (asc,) = self.options
+            if isinstance(self.options[0], str):
+                descending = self.options[0] == "descending"  # pragma: no cover
+            else:
+                descending, _ = self.options
             order = (
-                plc.types.Order.ASCENDING
-                if asc == "ascending"
-                else plc.types.Order.DESCENDING
+                plc.types.Order.DESCENDING if descending else plc.types.Order.ASCENDING
             )
             null_order = plc.types.NullOrder.BEFORE
             if column.null_count > 0 and (n := column.size) > 1:

python/cudf_polars/cudf_polars/dsl/ir.py

@@ -3187,19 +3187,21 @@ def do_evaluate(
 class HConcat(IR):
     """Concatenate dataframes horizontally."""
 
-    __slots__ = ("should_broadcast",)
-    _non_child = ("schema", "should_broadcast")
-    _n_non_child_args = 2
+    __slots__ = ("should_broadcast", "strict")
+    _non_child = ("schema", "should_broadcast", "strict")
+    _n_non_child_args = 3
 
     def __init__(
         self,
         schema: Schema,
         should_broadcast: bool,  # noqa: FBT001
+        strict: bool,  # noqa: FBT001
         *children: IR,
     ):
         self.schema = schema
         self.should_broadcast = should_broadcast
-        self._non_child_args = (schema, should_broadcast)
+        self.strict = strict
+        self._non_child_args = (schema, should_broadcast, strict)
         self.children = children
 
     @staticmethod
@@ -3242,6 +3244,7 @@ def do_evaluate(
         cls,
         schema: Schema,
         should_broadcast: bool,  # noqa: FBT001
+        strict: bool,  # noqa: FBT001
         *dfs: DataFrame,
         context: IRExecutionContext,
     ) -> DataFrame:
@@ -3259,6 +3262,13 @@ def do_evaluate(
                 result = DataFrame(ordered, stream=stream)
             else:
                 max_rows = max(df.num_rows for df in dfs)
+                if strict and any(df.num_rows != max_rows for df in dfs):
+                    heights = [df.num_rows for df in dfs]
+                    msg = (
+                        f"cannot concat DataFrames horizontally"
+                        f" with strict=True: height mismatch {heights}"
+                    )
+                    raise pl.exceptions.ShapeError(msg)
                 # Horizontal concatenation extends shorter tables with nulls
                 result = DataFrame(
                     itertools.chain.from_iterable(

python/cudf_polars/cudf_polars/dsl/translate.py

@@ -35,6 +35,7 @@
 from cudf_polars.utils.versions import (
     POLARS_VERSION_LT_136,
     POLARS_VERSION_LT_138,
+    POLARS_VERSION_LT_139,
 )
 
 if TYPE_CHECKING:
@@ -45,6 +46,29 @@
 __all__ = ["Translator", "translate_named_expr"]
 
 
+def _align_decimal_float_for_comparison(
+    *operands: expr.Expr,
+) -> tuple[expr.Expr, ...]:
+    """
+    Cast Decimal operands to Float64 when mixed with Float operands.
+
+    libcudf does not produce correct results for Decimal vs Float
+    comparisons. This matches Polars' supertype resolution where
+    Decimal + Float yields Float64.
+    """
+    has_decimal = any(plc.traits.is_fixed_point(op.dtype.plc_type) for op in operands)
+    has_float = any(plc.traits.is_floating_point(op.dtype.plc_type) for op in operands)
+    if has_decimal and has_float:
+        f64 = DataType(pl.Float64())
+        return tuple(
+            expr.Cast(f64, False, op)  # noqa: FBT003
+            if plc.traits.is_fixed_point(op.dtype.plc_type)
+            else op
+            for op in operands
+        )
+    return operands
+
+
 def _check_compression(data: bytes) -> str | None:
     # Vendored from Polars' SupportedCompression::check in
     # polars-io/src/utils/compression.rs
@@ -125,7 +149,7 @@ def translate_ir(self, *, n: int | None = None) -> ir.IR:
         # IR is versioned with major.minor, minor is bumped for backwards
         # compatible changes (e.g. adding new nodes), major is bumped for
         # incompatible changes (e.g. renaming nodes).
-        if (version := self.visitor.version()) >= (12, 1):
+        if (version := self.visitor.version()) >= (12, 2):
             e = NotImplementedError(
                 f"No support for polars IR {version=}"
             )  # pragma: no cover; no such version for now.
@@ -247,6 +271,17 @@ def __exit__(self, *args: Any) -> None:
         self.translator._expr_context = self._prev
 
 
+def _is_dynamic_pred(visitor: Any, expr_ir: Any) -> bool:
+    try:
+        visitor.view_expression(expr_ir.node)
+    except Exception as e:
+        if str(e) == "dynamic_pred":
+            return True
+        raise  # pragma: no cover
+    else:
+        return False
+
+
 @singledispatch
 def _translate_ir(node: Any, translator: Translator, schema: Schema) -> ir.IR:
     raise NotImplementedError(
@@ -307,7 +342,7 @@ def _(node: plrs._ir_nodes.Scan, translator: Translator, schema: Schema) -> ir.I
     if typ in ("csv", "ndjson"):
         for p in paths:
             if plc.io.SourceInfo._is_remote_uri(p):
-                continue
+                continue  # pragma: no cover
             data = _read_file_bytes(Path(p))
             compression = _check_compression(data)
             if compression is not None:
@@ -326,9 +361,12 @@ def _(node: plrs._ir_nodes.Scan, translator: Translator, schema: Schema) -> ir.I
         n_rows,
         row_index,
         include_file_paths,
-        translate_named_expr(translator, n=node.predicate, schema=schema)
-        if node.predicate is not None
-        else None,
+        (
+            None
+            if node.predicate is None
+            or _is_dynamic_pred(translator.visitor, node.predicate)
+            else translate_named_expr(translator, n=node.predicate, schema=schema)
+        ),
         parquet_options,
     )
 
@@ -537,6 +575,8 @@ 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):
+            return inp
         mask = translate_named_expr(translator, n=node.predicate, schema=inp.schema)
     return ir.Filter(schema, mask, inp)
 
@@ -585,9 +625,14 @@ def _(node: plrs._ir_nodes.Union, translator: Translator, schema: Schema) -> ir.
 
 @_translate_ir.register
 def _(node: plrs._ir_nodes.HConcat, translator: Translator, schema: Schema) -> ir.IR:
+    if POLARS_VERSION_LT_139:
+        strict = False  # pragma: no cover
+    else:
+        _, strict, _ = node.options
     return ir.HConcat(
         schema,
         False,  # noqa: FBT003
+        strict,
         *(translator.translate_ir(n=n) for n in node.inputs),
     )
 
@@ -625,7 +670,7 @@ def _(node: plrs._ir_nodes.Sink, translator: Translator, schema: Schema) -> ir.I
             if k in {"mkdir", "maintain_order", "sync_on_close"}:
                 options[k] = v
 
-    if sink_kind in ("Csv", "NDJson"):
+    if sink_kind in ("Csv", "NDJson", "Json"):
         compression = format_options.get("compression")
         if compression and compression != "Uncompressed":
             raise NotImplementedError(
@@ -748,6 +793,10 @@ def _(
             )
             (closed,) = options
             lop, rop = expr.BooleanFunction._BETWEEN_OPS[closed]
+            if not POLARS_VERSION_LT_139:
+                # IsBetween type coercion does not
+                # insert casts for Decimal/Float comparisons.
+                column, lo, hi = _align_decimal_float_for_comparison(column, lo, hi)
             return expr.BinOp(
                 dtype,
                 plc.binaryop.BinaryOperator.LOGICAL_AND,
@@ -820,14 +869,9 @@ def _(
     dtype: DataType,
     schema: Schema,
 ) -> expr.Expr:
-    if isinstance(
+    if hasattr(plrs._expr_nodes, "RollingGroupOptions") and isinstance(
         node.options, plrs._expr_nodes.RollingGroupOptions
-    ):  # pragma: no cover; polars gives Aexpr::Rolling node now
-        # TODO: As of polars 1.36.0, rolling is represented as a Rolling expression node
-        # but is currently not implemented in the python node visitor. Once we support it,
-        # we should move this branch to separate translator dispatch function for the new node.
-
-        # pl.col("a").rolling(...)
+    ):  # pragma: no cover; polars >=1.36 uses AExpr::Rolling now
         with set_expr_context(translator, ExecutionContext.ROLLING):
             agg = translator.translate_expr(n=node.function, schema=schema)
         name_generator = unique_names(schema)
@@ -928,6 +972,61 @@ def _(
     assert_never(node.options)
 
 
+def _translate_rolling(
+    node: plrs._expr_nodes.Rolling,
+    translator: Translator,
+    dtype: DataType,
+    schema: Schema,
+) -> expr.Expr:
+    # pl.col("a").rolling(...)
+    with set_expr_context(translator, ExecutionContext.ROLLING):
+        agg_expr = translator.translate_expr(n=node.function, schema=schema)
+    orderby = translator.visitor.view_expression(node.index_column).name
+    orderby_dtype = schema[orderby].plc_type
+    if plc.traits.is_integral(orderby_dtype):
+        orderby_dtype = plc.DataType(plc.TypeId.INT64)
+    name_generator = unique_names(schema)
+    aggs, named_post_agg = decompose_single_agg(
+        expr.NamedExpr(next(name_generator), agg_expr),
+        name_generator,
+        is_top=True,
+        context=ExecutionContext.ROLLING,
+    )
+    named_aggs = [a for a, _ in aggs]
+    closed_window = node.closed_window
+    if isinstance(named_post_agg.value, expr.Col):
+        (named_agg,) = named_aggs
+        return expr.RollingWindow(
+            named_agg.value.dtype,
+            orderby_dtype,
+            node.offset,
+            node.period,
+            closed_window,
+            orderby,
+            named_agg.value,
+        )
+    replacements: dict[expr.Expr, expr.Expr] = {
+        expr.Col(a.value.dtype, a.name): expr.RollingWindow(
+            a.value.dtype,
+            orderby_dtype,
+            node.offset,
+            node.period,
+            closed_window,
+            orderby,
+            a.value,
+        )
+        for a in named_aggs
+    }
+    return replace([named_post_agg.value], replacements)[0]
+
+
+# Rolling was added in polars 1.39. Use explicit registration (not annotation-based)
+# so that Python 3.14's stricter get_type_hints() resolution does not fail when
+# running on older polars versions that lack this node type.
+if hasattr(plrs._expr_nodes, "Rolling"):
+    _translate_expr.register(plrs._expr_nodes.Rolling)(_translate_rolling)
+
+
 @_translate_expr.register
 def _(
     node: plrs._expr_nodes.Literal,

python/cudf_polars/cudf_polars/experimental/expressions.py

@@ -506,6 +506,7 @@ def _decompose(
         input_ir = HConcat(
             schema,
             True,  # noqa: FBT003
+            False,  # noqa: FBT003
             *unique_input_irs,
         )
         partition_info[input_ir] = PartitionInfo(count=partition_count)

python/cudf_polars/cudf_polars/experimental/io.py

@@ -411,7 +411,7 @@ def _sink_to_file(
             # Path.open returns IO[Any] but SinkInfo needs more specific IO types
             sink = plc.io.types.SinkInfo([f])  # type: ignore[arg-type]
             Sink._write_csv(sink, use_options, df)
-    elif kind == "Json" if POLARS_VERSION_LT_137 else "NDJson":
+    elif kind == ("Json" if POLARS_VERSION_LT_137 else "NDJson"):
         mode = "wb" if writer_state is None else "ab"
         with Path.open(Path(path), mode) as f:
             # Path.open returns IO[Any] but SinkInfo needs more specific IO types