do not enforce meta for cum aggs

dask · Jan 10, 2024 · b21d4f2 · b21d4f2
1 parent 8119454
commit b21d4f2
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
@@ -3448,9 +3448,7 @@ def _describe_nonnumeric_1d(
         meta = data._meta_nonempty.describe(**datetime_is_numeric_kwarg)
         return new_dd_object(graph, name, meta, divisions=[None, None])
 
-    def _cum_agg(
-        self, op_name, chunk, aggregate, axis, skipna=True, chunk_kwargs=None, out=None
-    ):
+    def _cum_agg(self, op_name, chunk, aggregate, axis, skipna, chunk_kwargs, out):
         """Wrapper for cumulative operation"""
 
         axis = self._validate_axis(axis)
@@ -3463,7 +3461,12 @@ def _cum_agg(
             # cumulate each partitions
             name1 = f"{self._token_prefix}{op_name}-map"
             cumpart = map_partitions(
-                chunk, self, token=name1, meta=self, **chunk_kwargs
+                chunk,
+                self,
+                token=name1,
+                meta=self,
+                **chunk_kwargs,
+                enforce_metadata=False,
             )
 
             name2 = f"{self._token_prefix}{op_name}-take-last"
@@ -3473,6 +3476,7 @@ def _cum_agg(
                 skipna,
                 meta=meta_series_constructor(self)([], dtype="float"),
                 token=name2,
+                enforce_metadata=False,
             )
 
             suffix = tokenize(self)
@@ -7285,7 +7289,7 @@ def apply_and_enforce(*args, **kwargs):
     df = func(*args, **kwargs)
 
     if is_scalar(df) and is_series_like(meta):
-        df = type(meta)([df], dtype=meta.dtype)
+        df = type(meta)([df])
 
     if is_dataframe_like(df) or is_series_like(df) or is_index_like(df):
         if not len(df):