Repository: dask/dask
Pull request #11745 (2025-02-13 16:35:25)

Simplify assert_divisions
Small thing leading to https://github.com/dask/dask/pull/11736

I ran into an issue in this vicinity and this logic is replicated. Instead of using the `get(ddf.dask, ...)` it is now using the ordinary compute path. This also fixes an issue with `enforce_runtime_divisions` which was essentially not runnable before

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ±0       15 suites  ±0   4h 2m 27s ⏱️ - 9m 47s
 17 830 tests ±0   16 053 ✅  -   496   1 300 💤 + 19    477 ❌ +  477 
216 479 runs  ±0  180 090 ✅  - 5 160  31 432 💤 +203  4 957 ❌ +4 957 

For more details on these failures, see [this check](https://github.com/dask/dask/runs/37180712725).

Results for commit b6e3f8b5. ± Comparison against base commit c3f60178.

<details>
  <summary>This pull request <b>skips</b> 19 tests.</summary>

```
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-count-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-max-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-mean-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-min-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-prod-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-product-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-sem-None]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-sem-kwargs12]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-sem-kwargs13]
dask.dataframe.tests.test_arithmetics_reduction ‑ test_reductions_frame_dtypes[False-sem-kwargs14]
…
```
</details>

[test-results]:data:application/gzip;base64,H4sIALEnrmcC/02MOQ7DIBQFr2JRp2BfchkLvkFC8RJhqKzcPZAEO+XMe5oDhTj7Hd0HIm4D2kvMF00l2Ry3tTEXXFVV1/zZlWa487gXgCYlFuySj/hskuHrGGycq+PqbI0+pS1V106prC1OieTK/PiMa4wNPuU3zghntLseN0J19R+HbVliroCc9CxoJyYL0lnhDfbUUeOCoxA8sWC1tyAIer0B0ybagR8BAAA=


Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index 80df3cfbd..5506e5553 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -224,7 +224,7 @@ def convert_legacy_task(
             new_args.append(new)
         return Task(key, func, *new_args)
     try:
-        if isinstance(task, (bytes, int, float, str, tuple)):
+        if isinstance(task, (int, float, str, tuple)):
             if task in all_keys:
                 if key is None:
                     return Alias(task)
diff --git a/dask/core.py b/dask/core.py
index bf0c2dceb..dd61cc980 100644
--- a/dask/core.py
+++ b/dask/core.py
@@ -149,7 +149,7 @@ def keys_in_tasks(keys: Collection[Key], tasks: Iterable[Any], as_list: bool = F
 def iskey(key: object) -> bool:
     """Return True if the given object is a potential dask key; False otherwise.
 
-    The definition of a key in a Dask graph is any str, bytes, int, float, or tuple
+    The definition of a key in a Dask graph is any str, int, float, or tuple
     thereof.
 
     See Also
@@ -161,7 +161,7 @@ def iskey(key: object) -> bool:
     typ = type(key)
     if typ is tuple:
         return all(iskey(i) for i in cast(tuple, key))
-    return typ in {bytes, int, float, str}
+    return typ in {int, float, str}
 
 
 def validate_key(key: object) -> None:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index a0aa0d9fe..32d8bbcb6 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -18,6 +18,7 @@ from dask.dataframe.utils import (
     has_known_categories,
     is_dataframe_like,
     is_index_like,
+    is_scalar,
     is_series_like,
     meta_frame_constructor,
     meta_series_constructor,
@@ -95,7 +96,13 @@ def apply_and_enforce(*args, **kwargs):
     func = kwargs.pop("_func")
     meta = kwargs.pop("_meta")
     df = func(*args, **kwargs)
-    if is_dataframe_like(df) or is_series_like(df) or is_index_like(df):
+
+    if any(
+        bool(is_dataframe_like(obj) or is_series_like(obj) or is_index_like(obj))
+        for obj in [df, meta]
+    ):
+        if is_scalar(df):
+            df = pd.Series(df)
         if not len(df):
             return meta
         if is_dataframe_like(df):
diff --git a/dask/dataframe/dask_expr/_collection.py b/dask/dataframe/dask_expr/_collection.py
index 70a18d8a3..8335aba4d 100644
--- a/dask/dataframe/dask_expr/_collection.py
+++ b/dask/dataframe/dask_expr/_collection.py
@@ -106,7 +106,6 @@ from dask.dataframe.dask_expr._util import (
     _raise_if_object_series,
     _validate_axis,
     get_specified_shuffle,
-    is_scalar,
 )
 from dask.dataframe.dask_expr.io import FromPandasDivisions, FromScalars
 from dask.dataframe.dispatch import (
@@ -121,6 +120,7 @@ from dask.dataframe.utils import (
     has_known_categories,
     index_summary,
     insert_meta_param_description,
+    is_scalar,
     meta_frame_constructor,
     meta_series_constructor,
     pyarrow_strings_enabled,
diff --git a/dask/dataframe/dask_expr/_expr.py b/dask/dataframe/dask_expr/_expr.py
index 032717ba5..5216383f3 100644
--- a/dask/dataframe/dask_expr/_expr.py
+++ b/dask/dataframe/dask_expr/_expr.py
@@ -37,7 +37,6 @@ from dask.dataframe.dask_expr._util import (
     _calc_maybe_new_divisions,
     _convert_to_list,
     _tokenize_partial,
-    is_scalar,
 )
 from dask.dataframe.dispatch import make_meta, meta_nonempty
 from dask.dataframe.rolling import CombinedOutput, _head_timedelta, overlap_chunk
@@ -45,6 +44,7 @@ from dask.dataframe.shuffle import drop_overlap, get_overlap
 from dask.dataframe.utils import (
     clear_known_categories,
     drop_by_shallow_copy,
+    is_scalar,
     raise_on_meta_error,
     valid_divisions,
 )
@@ -1631,14 +1631,26 @@ class Where(Elemwise):
 
 
 def _check_divisions(df, i, division_min, division_max, last):
+    if not len(df):
+        return df
+    if is_index_like(df):
+        index = df
+    else:
+        try:
+            index = df.index.get_level_values(0)
+        except AttributeError:
+            index = df.index
     # Check divisions
-    real_min = df.index.min()
-    real_max = df.index.max()
+    real_min = index.min()
+    real_max = index.max()
     # Upper division of the last partition is often set to
     # the max value. For all other partitions, the upper
     # division should be greater than the maximum value.
-    valid_min = real_min >= division_min
-    valid_max = (real_max <= division_max) if last else (real_max < division_max)
+    valid_min = valid_max = True
+    if not pd.isna(division_min):
+        valid_min = real_min >= division_min
+    if not pd.isna(division_max):
+        valid_max = (real_max <= division_max) if last else (real_max < division_max)
     if not (valid_min and valid_max):
         raise RuntimeError(
             f"`enforce_runtime_divisions` failed for partition {i}."
@@ -1665,7 +1677,7 @@ class EnforceRuntimeDivisions(Blockwise):
             self.divisions[index + 1],
             index == (self.npartitions - 1),
         ]
-        return (self.operation,) + tuple(args)  # type: ignore
+        return Task(name, self.operation, *args)
 
 
 class Abs(Elemwise):
@@ -3867,7 +3879,9 @@ def plain_column_projection(expr, parent, dependents, additional_columns=None):
         # we are accesing the index
         column_union = []
 
-    if column_union == expr.frame.columns:
+    if column_union == expr.frame.columns or not column_union and expr.ndim < 2:
+        # this projection is for the index, but the elements are unknown, so
+        # don't project
         return
     result = type(expr)(expr.frame[column_union], *expr.operands[1:])
     if column_union == parent.operand("columns"):
diff --git a/dask/dataframe/dask_expr/_groupby.py b/dask/dataframe/dask_expr/_groupby.py
index d56cacd39..8c555c0e8 100644
--- a/dask/dataframe/dask_expr/_groupby.py
+++ b/dask/dataframe/dask_expr/_groupby.py
@@ -45,7 +45,6 @@ from dask.dataframe.dask_expr._util import (
     PANDAS_GE_300,
     _convert_to_list,
     get_specified_shuffle,
-    is_scalar,
 )
 from dask.dataframe.dispatch import concat, make_meta, meta_nonempty
 from dask.dataframe.groupby import (
@@ -82,7 +81,7 @@ from dask.dataframe.groupby import (
     _var_agg,
     _var_chunk,
 )
-from dask.dataframe.utils import insert_meta_param_description
+from dask.dataframe.utils import insert_meta_param_description, is_scalar
 from dask.typing import Key
 from dask.utils import M, derived_from, is_index_like
 
diff --git a/dask/dataframe/dask_expr/_indexing.py b/dask/dataframe/dask_expr/_indexing.py
index 12bd43f04..7758c032d 100644
--- a/dask/dataframe/dask_expr/_indexing.py
+++ b/dask/dataframe/dask_expr/_indexing.py
@@ -20,7 +20,6 @@ from dask.dataframe.dask_expr._expr import (
     Projection,
     are_co_aligned,
 )
-from dask.dataframe.dask_expr._util import is_scalar
 from dask.dataframe.dispatch import meta_nonempty
 from dask.dataframe.indexing import (
     _coerce_loc_index,
@@ -28,6 +27,7 @@ from dask.dataframe.indexing import (
     _partition_of_index_value,
     _partitions_of_index_values,
 )
+from dask.dataframe.utils import is_scalar
 from dask.typing import Key
 from dask.utils import is_arraylike, is_series_like
 
diff --git a/dask/dataframe/dask_expr/_merge.py b/dask/dataframe/dask_expr/_merge.py
index 190cde1e2..d6fecac69 100644
--- a/dask/dataframe/dask_expr/_merge.py
+++ b/dask/dataframe/dask_expr/_merge.py
@@ -33,7 +33,7 @@ from dask.dataframe.dask_expr._shuffle import (
     _is_numeric_cast_type,
     _select_columns_or_index,
 )
-from dask.dataframe.dask_expr._util import _convert_to_list, is_scalar
+from dask.dataframe.dask_expr._util import _convert_to_list
 from dask.dataframe.dispatch import make_meta, meta_nonempty
 from dask.dataframe.multi import (
     _concat_wrapper,
@@ -42,6 +42,7 @@ from dask.dataframe.multi import (
     merge_chunk,
 )
 from dask.dataframe.shuffle import partitioning_index
+from dask.dataframe.utils import is_scalar
 from dask.tokenize import _tokenize_deterministic
 from dask.typing import Key
 from dask.utils import apply, get_default_shuffle_method
diff --git a/dask/dataframe/dask_expr/_reductions.py b/dask/dataframe/dask_expr/_reductions.py
index 5ba667896..da9fb6a2a 100644
--- a/dask/dataframe/dask_expr/_reductions.py
+++ b/dask/dataframe/dask_expr/_reductions.py
@@ -38,8 +38,8 @@ from dask.dataframe.dask_expr._expr import (
     determine_column_projection,
     plain_column_projection,
 )
-from dask.dataframe.dask_expr._util import is_scalar
 from dask.dataframe.dispatch import make_meta, meta_nonempty
+from dask.dataframe.utils import is_scalar
 from dask.tokenize import _tokenize_deterministic
 from dask.typing import no_default
 from dask.utils import M, apply, funcname
diff --git a/dask/dataframe/dask_expr/_util.py b/dask/dataframe/dask_expr/_util.py
index aef4d8e90..c9de28dab 100644
--- a/dask/dataframe/dask_expr/_util.py
+++ b/dask/dataframe/dask_expr/_util.py
@@ -2,10 +2,9 @@ from __future__ import annotations
 
 import functools
 from collections import OrderedDict, UserDict
-from collections.abc import Hashable, Iterable, Sequence
+from collections.abc import Hashable
 from typing import Any, Literal, TypeVar, cast
 
-import numpy as np
 import pandas as pd
 from packaging.version import Version
 
@@ -79,22 +78,6 @@ def _convert_to_list(column) -> list | None:
     return column
 
 
-def is_scalar(x):
-    # np.isscalar does not work for some pandas scalars, for example pd.NA
-    if isinstance(x, (Sequence, Iterable)) and not isinstance(x, str):
-        return False
-    elif hasattr(x, "dtype"):
-        return isinstance(x, np.ScalarType)
-    if isinstance(x, dict):
-        return False
-    if isinstance(x, (str, int)) or x is None:
-        return True
-
-    from dask.dataframe.dask_expr._expr import Expr
-
-    return not isinstance(x, Expr)
-
-
 def _tokenize_partial(expr, ignore: list | None = None) -> str:
     # Helper function to "tokenize" the operands
     # that are not in the `ignore` list
diff --git a/dask/dataframe/dask_expr/diagnostics/_analyze.py b/dask/dataframe/dask_expr/diagnostics/_analyze.py
index 29d2cadc4..0dfd822ca 100644
--- a/dask/dataframe/dask_expr/diagnostics/_analyze.py
+++ b/dask/dataframe/dask_expr/diagnostics/_analyze.py
@@ -7,12 +7,12 @@ import pandas as pd
 
 from dask.base import DaskMethodsMixin
 from dask.dataframe.dask_expr._expr import Blockwise, Expr
-from dask.dataframe.dask_expr._util import is_scalar
 from dask.dataframe.dask_expr.diagnostics._explain import (
     _add_graphviz_edges,
     _explain_info,
 )
 from dask.dataframe.dask_expr.io.io import FusedIO
+from dask.dataframe.utils import is_scalar
 from dask.sizeof import sizeof
 from dask.tokenize import _tokenize_deterministic
 from dask.utils import format_bytes, import_required
diff --git a/dask/dataframe/dask_expr/io/io.py b/dask/dataframe/dask_expr/io/io.py
index e0edf672d..ab416156f 100644
--- a/dask/dataframe/dask_expr/io/io.py
+++ b/dask/dataframe/dask_expr/io/io.py
@@ -445,6 +445,8 @@ class FromPandas(PartitionsFiltered, BlockwiseIO):
             try:
                 return list(self.frame.columns)
             except AttributeError:
+                if self.ndim == 1:
+                    return [self.name]
                 return []
         else:
             return _convert_to_list(columns_operand)
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index 1911e5de7..7260ae0d6 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -5,7 +5,7 @@ import re
 import sys
 import textwrap
 import traceback
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from contextlib import contextmanager
 from numbers import Number
 from typing import TypeVar, overload
@@ -15,7 +15,7 @@ import pandas as pd
 from pandas.api.types import is_dtype_equal
 
 import dask
-from dask.base import get_scheduler, is_dask_collection
+from dask.base import is_dask_collection
 from dask.core import get_deps
 from dask.dataframe._compat import tm  # noqa: F401
 from dask.dataframe.dispatch import (  # noqa : F401
@@ -43,6 +43,22 @@ except ImportError:
     pass
 
 
+def is_scalar(x):
+    # np.isscalar does not work for some pandas scalars, for example pd.NA
+    if isinstance(x, (Sequence, Iterable)) and not isinstance(x, str):
+        return False
+    elif hasattr(x, "dtype"):
+        return isinstance(x, np.ScalarType)
+    if isinstance(x, dict):
+        return False
+    if isinstance(x, (str, int)) or x is None:
+        return True
+
+    from dask.dataframe.dask_expr._expr import Expr
+
+    return not isinstance(x, Expr)
+
+
 def is_integer_na_dtype(t):
     dtype = getattr(t, "dtype", t)
     types = (
@@ -587,24 +603,7 @@ def assert_divisions(ddf, scheduler=None):
     if not getattr(ddf, "known_divisions", False):
         return
 
-    def index(x):
-        if is_index_like(x):
-            return x
-        try:
-            return x.index.get_level_values(0)
-        except AttributeError:
-            return x.index
-
-    get = get_scheduler(scheduler=scheduler, collections=[type(ddf)])
-    results = get(ddf.dask, ddf.__dask_keys__())
-    for i, df in enumerate(results[:-1]):
-        if len(df):
-            assert index(df).min() >= ddf.divisions[i]
-            assert index(df).max() < ddf.divisions[i + 1]
-
-    if len(results[-1]):
-        assert index(results[-1]).min() >= ddf.divisions[-2]
-        assert index(results[-1]).max() <= ddf.divisions[-1]
+    ddf.enforce_runtime_divisions().compute(scheduler=scheduler)
 
 
 def assert_sane_keynames(ddf):
diff --git a/dask/typing.py b/dask/typing.py
index 6c67ff746..8a970f87e 100644
--- a/dask/typing.py
+++ b/dask/typing.py
@@ -25,7 +25,7 @@ CollType_co = TypeVar("CollType_co", bound="DaskCollection", covariant=True)
 PostComputeCallable = Callable
 
 
-Key: TypeAlias = Union[str, bytes, int, float, tuple["Key", ...]]
+Key: TypeAlias = Union[str, int, float, tuple["Key", ...]]
 # FIXME: This type is a little misleading. Low level graphs are often
 # MutableMappings but HLGs are not
 Graph: TypeAlias = Mapping[Key, Any]
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index fbb79cf08..92b1da12d 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -5,19 +5,68 @@ Changelog
 
     This is not exhaustive. For an exhaustive list of changes, see the git log.
 
-.. _v2025.1.0:
+.. _v2025.2.0:
 
 2025.2.0
+--------
 
 Highlights
 ^^^^^^^^^^
-
 This release includes a critical fix that fixes a deadlock that can arise
 when seceded task are rescheduled, or cancelled and resubmitted, e.g. due
 to a worker being lost.
 
 See :pr-distributed:`8991` by `Hendrik Makait`_ for more details.
 
+.. dropdown:: Additional changes
+
+  - Add big array example (:pr:`11744`) `James Bourbeau`_
+  - Fix exploding chunksizes in pad for constant padding (:pr:`11743`) `Patrick Hoefler`_
+  - Move optimize method to base class (:pr:`11742`) `Florian Jetter`_
+  - Add changelog entry for fixed deadlock (:pr:`11741`) `Hendrik Makait`_
+  - Fix graph creation in ``dask-expr`` ``to_delayed`` (:pr:`11739`) `Patrick Hoefler`_
+  - Remove culling from delayed optimisation (:pr:`11737`) `Patrick Hoefler`_
+  - Compute meta for from_map on the cluster (:pr:`11738`) `Patrick Hoefler`_
+  - Bugs in ``__setitem__`` with dask bool mask (:pr:`11728`) `Guido Imperiale`_
+  - Implement infrastructure, random, blockwise and Elemwise  (:pr:`11689`) `Patrick Hoefler`_
+  - ``array`` / ``asarray`` with both ``like=`` and ``dtype=`` (:pr:`11733`) `Guido Imperiale`_
+  - Fix annotations warnings test (:pr:`11734`) `Patrick Hoefler`_
+  - Catch warnings when writing to remote storage with to_parquet (:pr:`11731`) `Patrick Hoefler`_
+  - Remove LocalCluster from tests (:pr:`11729`) `Patrick Hoefler`_
+  - Fix partition pruning when using from_array (:pr:`11725`) `Patrick Hoefler`_
+  - Fix concatentation with mixed dtype columns (:pr:`11727`) `Patrick Hoefler`_
+  - ``arange``: fix extreme values (:pr:`11707`) `Guido Imperiale`_
+  - Graph corruption on scalar ``getitem`` -> ``setitem`` (:pr:`11723`) `Guido Imperiale`_
+  - Never share buffers after compute() (:pr:`11697`) `Guido Imperiale`_
+  - Extract Dask Array from xarray DataArray in from_array (:pr:`11712`) `Patrick Hoefler`_
+  - ``arange``: support kwargs (:pr:`11710`) `Guido Imperiale`_
+  - Ensure ``normalize_token`` is threadsafe (:pr:`11709`) `Florian Jetter`_
+  - Expand advise for instance types and processes (:pr:`11705`) `Florian Jetter`_
+  - Drop legacy timeseries implementation (:pr:`11704`) `Florian Jetter`_
+  - Update Dask Cloud Provider documentation to include Nebius as a supported cloud option (:pr:`11703`) `Alexander`_
+  - Fix ``normalize_chunks`` when squashing into a single chunk (:pr:`11702`) `Patrick Hoefler`_
+  - Fix positional indexing with ``newaxis`` (:pr:`11699`) `Patrick Hoefler`_
+  - Set array backend in scipy-sparse-indexing (:pr:`11700`) `Tom Augspurger`_
+  - Fix ``value_counts`` shuffling strategy (:pr:`11698`) `Patrick Hoefler`_
+  - Disentangle core expression class from dataframe specific code (:pr:`11688`) `Patrick Hoefler`_
+  - Bump ``conda-incubator/setup-miniconda`` from 3.1.0 to 3.1.1 (:pr:`11685`)
+  - Fixup dataframe conversion from array methods (:pr:`11684`) `Patrick Hoefler`_
+  - Remove remaining artifacts of ``fastparquet`` (:pr:`11682`) `Patrick Hoefler`_
+
+  - Remove traceback from ``sizeof`` failure warning (:pr-distributed:`9006`) `Jacob Tomlinson`_
+  - Hotfix: Ignore negative occupancy (:pr-distributed:`9012`) `Hendrik Makait`_
+  - Remove expensive tokenization for key uniqueness check (:pr-distributed:`9009`) `Patrick Hoefler`_
+  - Fix CI for changes in ``from_map`` (:pr-distributed:`9011`) `Patrick Hoefler`_
+  - Avoid handling stale long-running messages on scheduler (:pr-distributed:`8991`) `Hendrik Makait`_
+  - Bump ``test_stress`` timeout (:pr-distributed:`9002`) `Tom Augspurger`_
+  - Poll in ``test_rmm_metrics`` test (:pr-distributed:`9004`) `Tom Augspurger`_
+  - Cache occupancy in ``WorkStealing.balance()`` (:pr-distributed:`9005`) `Hendrik Makait`_
+  - Homogeneous balancing by accounting for in-flight requests (:pr-distributed:`9003`) `Hendrik Makait`_
+  - Consistent estimation of task duration between stealing, adaptive and occupancy calculation (:pr-distributed:`9000`) `Hendrik Makait`_
+  - Increase default work-stealing interval by 10x (:pr-distributed:`8997`) `Hendrik Makait`_
+  - Remove occupancy plot from status dashboard (:pr-distributed:`8995`) `Hendrik Makait`_
+  - Bump ``conda-incubator/setup-miniconda`` from 3.1.0 to 3.1.1 (:pr-distributed:`8990`)
+
 
 .. _v2025.1.0:
 
@@ -9220,3 +9269,5 @@ Other
 .. _`Ilan Gold`: https://github.com/ilan-gold
 .. _`Jean-Baptiste Bayle`: https://github.com/j2bbayle
 .. _`dchudz`: https://github.com/dchudz
+.. _`Guido Imperiale`: https://github.com/crusaderky
+.. _`Alexander`: https://github.com/SalikovAlex
diff --git a/pyproject.toml b/pyproject.toml
index 1e8a62480..6548c4bc1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -55,7 +55,7 @@ dataframe = [
     "pandas >= 2.0",
     "pyarrow>=14.0.1",
 ]
-distributed = ["distributed == 2025.1.0"]
+distributed = ["distributed == 2025.2.0"]
 diagnostics = [
     "bokeh >= 3.1.0",
     "jinja2 >= 2.10.3",


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 958c6e13e..f086d4405 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -125,7 +125,8 @@ jobs:
         # ensure this runs even if pytest fails
         if: >
           always() &&
-          (steps.run_tests.outcome == 'success' || steps.run_tests.outcome == 'failure')
+          (steps.run_tests.outcome == 'success' || steps.run_tests.outcome == 'failure') &&
+          matrix.array-expr == 'false'
         uses: actions/upload-artifact@v4
         with:
           name: ${{ matrix.os }}-${{ matrix.environment }}-${{ matrix.extra }}
diff --git a/dask/dataframe/dask_expr/tests/test_collection.py b/dask/dataframe/dask_expr/tests/test_collection.py
index a4d2d15c4..af5bba927 100644
--- a/dask/dataframe/dask_expr/tests/test_collection.py
+++ b/dask/dataframe/dask_expr/tests/test_collection.py
@@ -22,7 +22,6 @@ from dask.dataframe.dask_expr import (
     expr,
     from_dict,
     from_pandas,
-    is_scalar,
     isna,
     optimize,
     to_datetime,
@@ -40,7 +39,7 @@ from dask.dataframe.dask_expr._shuffle import Shuffle
 from dask.dataframe.dask_expr.datasets import timeseries
 from dask.dataframe.dask_expr.io import FromPandas
 from dask.dataframe.dask_expr.tests._util import _backend_library, assert_eq, xfail_gpu
-from dask.dataframe.utils import UNKNOWN_CATEGORIES, pyarrow_strings_enabled
+from dask.dataframe.utils import UNKNOWN_CATEGORIES, is_scalar, pyarrow_strings_enabled
 from dask.utils import M
 
 # Set DataFrame backend for this module
@@ -2785,3 +2784,11 @@ def test_to_backend_simplify():
         assert str(df2.expr) != str(df[["y"]].expr)
         df3 = df2.simplify()
         assert str(df3.expr) == str(df[["y"]].expr)
+
+
+def test_projection_on_series():
+    pdf = pd.DataFrame(data={"a": [1, 3, 2]}).a
+    df = from_pandas(pdf, npartitions=1)
+    result = df.replace(1, 5)
+    assert_eq(result, pdf.replace(1, 5))
+    assert_eq(result.fillna(0), pdf.replace(1, 5).fillna(0))
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index e9716e420..52e8e8d15 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -903,6 +903,7 @@ def test_map_partitions():
         d.map_partitions(lambda df: 1),
         pd.Series([1, 1, 1], dtype=np.int64),
         check_divisions=False,
+        check_index=False,
     )
 
 
@@ -5353,3 +5354,19 @@ def test_array_to_df_conversion():
     result = arr.map_blocks(foo, meta=pd.DataFrame(np.random.random((1, 10))))
     expected = pd.DataFrame(arr.compute())
     assert_eq(result, expected, check_index=False)
+
+
+@pytest.mark.parametrize("npartitions", [1, 5])
+def test_map_partitions_always_series(npartitions):
+    pdf = pd.DataFrame({"x": range(50)})
+    ddf = dd.from_pandas(pdf, npartitions)
+
+    def assert_series(x):
+        assert isinstance(x, pd.Series)
+        return x
+
+    res = ddf.x.map_partitions(M.min).map_partitions(assert_series).compute()
+    assert len(res) == ddf.npartitions
+    assert min(res) == min(pdf.x)
+
+    assert ddf.x.map_partitions(M.min).count().compute() == ddf.npartitions