Merge branch-25.06 into branch-25.08

Author: Matt711

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

@@ -66,7 +66,7 @@ dependencies:
 - pandas
 - pandas>=2.0,<2.2.4dev0
 - pandoc
-- polars>=1.24,<1.28
+- polars>=1.25,<1.29
 - pre-commit
 - ptxcompiler
 - pyarrow>=14.0.0,<20.0.0a0

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

@@ -68,7 +68,7 @@ dependencies:
 - pandas
 - pandas>=2.0,<2.2.4dev0
 - pandoc
-- polars>=1.24,<1.28
+- polars>=1.25,<1.29
 - pre-commit
 - ptxcompiler
 - pyarrow>=14.0.0,<20.0.0a0

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

@@ -65,7 +65,7 @@ dependencies:
 - pandas
 - pandas>=2.0,<2.2.4dev0
 - pandoc
-- polars>=1.24,<1.28
+- polars>=1.25,<1.29
 - pre-commit
 - pyarrow>=14.0.0,<20.0.0a0
 - pydata-sphinx-theme>=0.15.4

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

@@ -66,7 +66,7 @@ dependencies:
 - pandas
 - pandas>=2.0,<2.2.4dev0
 - pandoc
-- polars>=1.24,<1.28
+- polars>=1.25,<1.29
 - pre-commit
 - pyarrow>=14.0.0,<20.0.0a0
 - pydata-sphinx-theme>=0.15.4

dependencies.yaml

@@ -831,7 +831,7 @@ dependencies:
     common:
       - output_types: [conda, requirements, pyproject]
         packages:
-          - polars>=1.24,<1.28
+          - polars>=1.25,<1.29
   run_cudf_polars_experimental:
     common:
       - output_types: [conda, requirements, pyproject]

python/cudf_polars/cudf_polars/callback.py

@@ -23,7 +23,6 @@
 
 from cudf_polars.dsl.translate import Translator
 from cudf_polars.utils.timer import Timer
-from cudf_polars.utils.versions import POLARS_VERSION_LT_125
 
 if TYPE_CHECKING:
     from collections.abc import Generator
@@ -308,28 +307,12 @@ def execute_with_cudf(
             if translator.config_options.raise_on_fail:
                 raise exception
         else:
-            if POLARS_VERSION_LT_125:  # pragma: no cover
-                nt.set_udf(
-                    partial(
-                        _callback,
-                        ir,
-                        should_time=False,
-                        memory_resource=memory_resource,
-                        config_options=translator.config_options,
-                        timer=None,
-                    )
+            nt.set_udf(
+                partial(
+                    _callback,
+                    ir,
+                    memory_resource=memory_resource,
+                    config_options=translator.config_options,
+                    timer=timer,
                 )
-            else:
-                nt.set_udf(
-                    partial(
-                        _callback,
-                        ir,
-                        memory_resource=memory_resource,
-                        config_options=translator.config_options,
-                        timer=timer,
-                    )
-                )
-
-
-if POLARS_VERSION_LT_125:  # pragma: no cover
-    execute_with_cudf = partial(execute_with_cudf, duration_since_start=None)
+            )

python/cudf_polars/cudf_polars/containers/dataframe.py

@@ -18,7 +18,7 @@
 if TYPE_CHECKING:
     from collections.abc import Iterable, Mapping, Sequence, Set
 
-    from typing_extensions import Self
+    from typing_extensions import Any, Self
 
     from cudf_polars.typing import ColumnOptions, DataFrameHeader, Slice
 
@@ -270,7 +270,7 @@ def discard_columns(self, names: Set[str]) -> Self:
         """Drop columns by name."""
         return type(self)(column for column in self.columns if column.name not in names)
 
-    def select(self, names: Sequence[str]) -> Self:
+    def select(self, names: Sequence[str] | Mapping[str, Any]) -> Self:
         """Select columns by name returning DataFrame."""
         try:
             return type(self)(self.column_map[name] for name in names)

python/cudf_polars/cudf_polars/dsl/expressions/boolean.py

@@ -19,6 +19,7 @@
     ExecutionContext,
     Expr,
 )
+from cudf_polars.utils.versions import POLARS_VERSION_LT_128
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -87,9 +88,11 @@ def __init__(
             BooleanFunction.Name.IsLastDistinct,
             BooleanFunction.Name.IsUnique,
         )
-        if self.name is BooleanFunction.Name.IsIn and not all(
-            c.dtype == self.children[0].dtype for c in self.children
-        ):
+        if (
+            POLARS_VERSION_LT_128
+            and self.name is BooleanFunction.Name.IsIn
+            and not all(c.dtype == self.children[0].dtype for c in self.children)
+        ):  # pragma: no cover
             # TODO: If polars IR doesn't put the casts in, we need to
             # mimic the supertype promotion rules.
             raise NotImplementedError("IsIn doesn't support supertype casting")

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

@@ -13,6 +13,7 @@
 from cudf_polars.dsl.expressions.base import ExecutionContext, Expr
 from cudf_polars.dsl.expressions.literal import Literal
 from cudf_polars.utils import dtypes
+from cudf_polars.utils.versions import POLARS_VERSION_LT_128
 
 if TYPE_CHECKING:
     from cudf_polars.containers import DataFrame
@@ -233,6 +234,14 @@ def do_evaluate(
             else:
                 evaluated = self.children[1].evaluate(df, context=context)
                 arg = evaluated.obj_scalar if evaluated.is_scalar else evaluated.obj
+            if (
+                not POLARS_VERSION_LT_128
+                and isinstance(arg, plc.Scalar)
+                and dtypes.can_cast(column.obj.type(), arg.type())
+            ):  # pragma: no cover
+                arg = plc.unary.cast(
+                    plc.Column.from_scalar(arg, 1), column.obj.type()
+                ).to_scalar()
             return Column(plc.replace.replace_nulls(column.obj, arg))
         elif self.name in self._OP_MAPPING:
             column = self.children[0].evaluate(df, context=context)

python/cudf_polars/cudf_polars/dsl/ir.py

@@ -33,6 +33,7 @@
 from cudf_polars.dsl.nodebase import Node
 from cudf_polars.dsl.to_ast import to_ast, to_parquet_filter
 from cudf_polars.utils import dtypes
+from cudf_polars.utils.versions import POLARS_VERSION_LT_128
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Hashable, Iterable, Sequence
@@ -373,7 +374,9 @@ def __init__(
             # TODO: polars has this implemented for parquet,
             # maybe we can do this too?
             raise NotImplementedError("slice pushdown for negative slices")
-        if self.typ in {"csv"} and self.skip_rows != 0:  # pragma: no cover
+        if (
+            POLARS_VERSION_LT_128 and self.typ in {"csv"} and self.skip_rows != 0
+        ):  # pragma: no cover
             # This comes from slice pushdown, but that
             # optimization doesn't happen right now
             raise NotImplementedError("skipping rows in CSV reader")
@@ -383,7 +386,7 @@ def __init__(
             raise NotImplementedError(
                 "Read from cloud storage"
             )  # pragma: no cover; no test yet
-        if any(p.startswith("https://") for p in self.paths):
+        if any(str(p).startswith("https:/") for p in self.paths):
             raise NotImplementedError("Read from https")
         if self.typ == "csv":
             if self.reader_options["skip_rows_after_header"] != 0:
@@ -459,7 +462,8 @@ def add_file_paths(
         Each path is repeated according to the number of rows read from it.
         """
         (filepaths,) = plc.filling.repeat(
-            plc.Table([plc.interop.from_arrow(pa.array(paths))]),
+            # TODO: Remove call from_arrow when we support python list to Column
+            plc.Table([plc.interop.from_arrow(pa.array(map(str, paths)))]),
             plc.interop.from_arrow(pa.array(rows_per_path, type=pa.int32())),
         ).columns()
         return df.with_columns([Column(filepaths, name=name)])
@@ -481,6 +485,17 @@ def do_evaluate(
     ) -> DataFrame:
         """Evaluate and return a dataframe."""
         if typ == "csv":
+
+            def read_csv_header(
+                path: Path | str, sep: str
+            ) -> list[str]:  # pragma: no cover
+                with Path(path).open() as f:
+                    for line in f:
+                        stripped = line.strip()
+                        if stripped:
+                            return stripped.split(sep)
+                return []
+
             parse_options = reader_options["parse_options"]
             sep = chr(parse_options["separator"])
             quote = chr(parse_options["quote_char"])
@@ -524,7 +539,9 @@ def do_evaluate(
                 options = (
                     plc.io.csv.CsvReaderOptions.builder(plc.io.SourceInfo([path]))
                     .nrows(n_rows)
-                    .skiprows(skiprows)
+                    .skiprows(
+                        skiprows if POLARS_VERSION_LT_128 else skiprows + skip_rows
+                    )  # pragma: no cover
                     .lineterminator(str(eol))
                     .quotechar(str(quote))
                     .decimal(decimal)
@@ -535,6 +552,13 @@ def do_evaluate(
                 options.set_delimiter(str(sep))
                 if column_names is not None:
                     options.set_names([str(name) for name in column_names])
+                else:
+                    if (
+                        not POLARS_VERSION_LT_128 and skip_rows > header
+                    ):  # pragma: no cover
+                        # We need to read the header otherwise we would skip it
+                        column_names = read_csv_header(path, str(sep))
+                        options.set_names(column_names)
                 options.set_header(header)
                 options.set_dtypes(schema)
                 if usecols is not None:
@@ -691,6 +715,8 @@ def slice_skip(tbl: plc.Table) -> plc.Table:
                 name=name,
             )
             df = DataFrame([index_col, *df.columns])
+            if next(iter(schema)) != name:
+                df = df.select(schema)
         assert all(c.obj.type() == schema[name] for name, c in df.column_map.items())
         if predicate is None:
             return df