Repository: dask/dask

Issue:
`.value_counts()` failed to aggregate across partitions
<!-- Please include a self-contained copy-pastable example that generates the issue if possible.

Please be concise with code posted. See guidelines below on how to provide a good bug report:

- Craft Minimal Bug Reports http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports
- Minimal Complete Verifiable Examples https://stackoverflow.com/help/mcve

Bug reports that follow these guidelines are easier to diagnose, and so are often handled much more quickly.
-->

**Describe the issue**: In some cases, `.value_counts()` fails to provide *unique* counts (i.e., does not aggregate across partitions).

**Minimal Complete Verifiable Example**:

Code
```python
import pandas as pd
import dask.dataframe as dd

# Attempt 1: as expected
df1 = pd.DataFrame({"A": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                    "B": [11, 2, 13, 4, 15, 6, 17, 8, 19, 10]})
ddf1 = dd.from_pandas(df1, npartitions=2)

print((ddf1["A"] == ddf1["B"]).value_counts(dropna=False).compute())
print((ddf1["A"] == ddf1["B"]).compute().value_counts(dropna=False))

# Attempt 2: problematic
df2 = pd.DataFrame({"A": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],  # Same as `df1`, but `2` swapped for `12` in `B`
                    "B": [11, 12, 13, 4, 15, 6, 17, 8, 19, 10]})
ddf2 = dd.from_pandas(df2, npartitions=2)

print((ddf2["A"] == ddf2["B"]).value_counts(dropna=False).compute())
print((ddf2["A"] == ddf2["B"]).compute().value_counts(dropna=False))
```

Output
```
False    5
True     5
Name: count, dtype: int64
False    5
True     5
Name: count, dtype: int64
False    2
True     3
False    4
True     1
Name: count, dtype: int64
False    6
True     4
Name: count, dtype: int64
```

**Anything else we need to know?**: N/A

**Environment**:

- Dask version: 2025.1.0
- Python version: 3.10.16
- Operating System: Ubuntu 24.04.1 LTS
- Install method (conda, pip, source): pip


Discussion:
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 25m 56s ⏱️ + 5m 51s
 17 142 tests + 1   15 962 ✅ + 1   1 180 💤 ±0  0 ❌ ±0 
211 224 runs  +12  194 201 ✅ +14  17 023 💤  - 2  0 ❌ ±0 

Results for commit 9fccdaac. ± Comparison against base commit d8eb38e9.

[test-results]:data:application/gzip;base64,H4sIANx5k2cC/02NzQ6DIBAGX8Vw7oFdV4G+TENXTEhVGoRT03cvWv+OM99m9iN6P7hZ3CtobpWYs08ndTna5MO0smnaosqa1l0B4c6POTP/j9qLfPn3IkHLw/XWD8WdwsUY4mZinpY0AiDSxkfaEEo45JZWEuvdXdIrX8scxtGnAsL0zJ21DPAktKoxqiNgx1KTQcTaOUXalZffH7q8MK0aAQAA


Fix (patch):
diff --git a/dask/dataframe/dask_expr/_reductions.py b/dask/dataframe/dask_expr/_reductions.py
index 5b5f248dd..5ba667896 100644
--- a/dask/dataframe/dask_expr/_reductions.py
+++ b/dask/dataframe/dask_expr/_reductions.py
@@ -1427,6 +1427,7 @@ class ValueCounts(ReductionConstantDim):
     reduction_chunk = M.value_counts
     reduction_aggregate = methods.value_counts_aggregate
     reduction_combine = methods.value_counts_combine
+    split_by = None
 
     @functools.cached_property
     def _meta(self):
@@ -1441,8 +1442,8 @@ class ValueCounts(ReductionConstantDim):
             return func(_concat(inputs), observed=True, **kwargs)
 
     @property
-    def split_by(self):
-        return self.frame._meta.name
+    def shuffle_by_index(self):
+        return True
 
     @property
     def chunk_kwargs(self):


Test changes:
diff --git a/dask/dataframe/dask_expr/tests/test_reductions.py b/dask/dataframe/dask_expr/tests/test_reductions.py
index b10030ac0..744e648cb 100644
--- a/dask/dataframe/dask_expr/tests/test_reductions.py
+++ b/dask/dataframe/dask_expr/tests/test_reductions.py
@@ -558,3 +558,16 @@ def test_reductions_timestamps_display():
     data = pd.to_datetime(["2024-10-02 12:00:00", "2024-10-02 14:00:00"])
     df = from_pandas(pd.DataFrame({"valid_time": data}))
     assert df["valid_time"].min().__repr__()
+
+
+def test_value_counts_shuffle_properly():
+    pdf = pd.DataFrame(
+        {
+            "A": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
+            "B": [11, 12, 13, 4, 15, 6, 17, 8, 19, 10],
+        }
+    )
+    df = from_pandas(pdf, npartitions=2)
+    result = (df["A"] == df["B"]).value_counts()
+    expected = (pdf["A"] == pdf["B"]).value_counts()
+    assert_eq(result, expected)