Add cudf support to to_datetime and _maybe_from_pandas

dask_expr/_expr.py

@@ -44,6 +44,7 @@
     M,
     apply,
     funcname,
+    get_meta_library,
     has_keyword,
     is_arraylike,
     partial_by_order,
@@ -1448,7 +1449,10 @@ class ToDatetime(Elemwise):
     _parameters = ["frame", "kwargs", "meta"]
     _defaults = {"kwargs": None}
     _keyword_only = ["kwargs", "meta"]
-    operation = staticmethod(pd.to_datetime)
+
+    @staticmethod
+    def operation(*args, **kwargs):
+        return get_meta_library(args[0]).to_datetime(*args, **kwargs)
 
     @functools.cached_property
     def _kwargs(self):

dask_expr/_util.py

@@ -11,6 +11,7 @@
 from dask import config
 from dask.base import normalize_token, tokenize
 from dask.dataframe._compat import is_string_dtype
+from dask.dataframe.core import is_dask_collection, is_dataframe_like, is_series_like
 from dask.utils import get_default_shuffle_method
 from packaging.version import Version
 
@@ -173,10 +174,11 @@ def normalize_data_wrapper(data):
 def _maybe_from_pandas(dfs):
     from dask_expr import from_pandas
 
-    dfs = [
-        from_pandas(df, 1) if isinstance(df, (pd.Series, pd.DataFrame)) else df
-        for df in dfs
-    ]
+    def _pd_series_or_dataframe(x):
+        # `x` can be a cudf Series/DataFrame
+        return not is_dask_collection(x) and (is_series_like(x) or is_dataframe_like(x))
+
+    dfs = [from_pandas(df, 1) if _pd_series_or_dataframe(df) else df for df in dfs]
     return dfs