[Data] map-groupby-aggregate seems to hang indefinitely on scheduler

[HenryL27]

What happened + What you expected to happen

I was writing some clustering code using ray (pared-down repro script below), and ray hangs on it.

When I run this on ray 2.50 (1-laptop cluster), it seems to hang forever with messages like:

(autoscaler +25s) Warning: The following resource request cannot be scheduled right now: {'CPU': 0.25, 'memory': 1073741824.0}. This is likely due to all cluster resources being claimed by actors. Consider creating fewer actors or adding more nodes to this Ray cluster

Which seems like an awful lot of memory to be requesting. I also have this plan log thingy

- HashAggregate(key_columns=('cluster',), num_partitions=200): Tasks: 15 [backpressured:tasks]; Actors: 0; Queued blocks: 67; Resources: 10.0 CPU, 0.0B object store: : 0.00 row [01:05, ? row/s]                                                                                                                                                                                         *- Shuffle: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 118/118 [01:05<00:00, 1.80 row/s]
  *- Aggregation: : 0.00 row [01:05, ? row/s]

Where it seems like 200 is the wrong number of partitions to default to? And if each partition is trying to create a HashAggregate actor using 10GB of ram I can see how that would break!

Result of this is the job hangs forever, asking me to connect more nodes to my cluster (but I'm just running on my laptop)

Notes: in 2.47 this doesn't seem to be an issue; also if I remove the map (set the cluster value when constructing the rows) it also seems to be ok; also if I set num_partitions in the groupby to something reasonable myself, I'm ok.

Versions / Dependencies

ray: 2.50.0
python: 3.12.0 (CPython)
os: Macos 26.0.1 (M2 Pro, 16GB Ram)

Reproduction script

import numpy as np
from ray.data import from_items
from ray.data.aggregate import AggregateFn
import random


def pc(row):
    row["cluster"] = random.randint(0, 10)
    return row


update_centroids = AggregateFn(
    init=lambda v: {"v": [0] * 10, "c": 0},
    accumulate_row=lambda a, row: {"v": [x + y for x, y in zip(a["v"], row["vector"])], "c": a["c"] + 1},
    merge=lambda a1, a2: {"v": [x + y for x, y in zip(a1["v"], a2["v"])], "c": a1["c"] + a2["c"]},
    name="centroids",
)


items = [{"vector": np.random.uniform(0, 1, (10)), "cluster": random.randint(0, 10)} for _ in range(200)]
ds = from_items(items)

initial_centroids = [i["vector"] for i in items[:10]]

ds.map(pc).groupby("cluster").aggregate(update_centroids).take()

Issue Severity

High: It blocks me from completing my task.

[iamjustinhsu]

Hi @HenryL27! Thanks for pointing this out! This is definitely unexpected behavior on our side, and we are gonna to fix it. The reason why it is failing is because the env var RAY_DATA_DEFAULT_HASH_SHUFFLE_AGGREGATOR_MEMORY_ALLOCATION is default set to 1GiB, and that isn't a correct estimation. If you set it to something lower, the pipeline will succeed.

EDIT: Although, removing the map is also peculiar, so will also take a look into that
EDIT2: Ok maybe I spoke too soon, it's a little unclear why

[iamjustinhsu]

@HenryL27, is it deterministically failing all the time?

[HenryL27]

It usually gets almost all the way through the job and then gets stuck. I confess the longest I've let it go like this is about 10 minutes so it might unstick itself, but it's a halting problem, ya know?

But yeah, without the tweaks I mentioned in notes I've not seen it complete

[HenryL27]

@iamjustinhsu did some more futzing, tried an AggregateFnV2 in case that mattered somehow, but that also broke in the same way

class update_centroids_v2(AggregateFnV2):
    def __init__(self):
        super().__init__("update_centroids_v2", lambda: {"v": [0] * 10, "c": 0}, on=None, ignore_nulls=True)

    def aggregate_block(self, block):
        return {"v": block["vector"].to_numpy().sum(), "c": len(block)}

    def combine(self, r1, r2):
        return {"v": r1["v"] + r2["v"], "c": r1["c"] + r2["c"]}

    def finalize(self, r):
        return r["v"] / r["c"]

[iamjustinhsu]

hmm, it's a bit difficult to reproduce because it's non-deterministic. I did encounter it a couple of times, but now the problem seems to have gone away on my laptop ... When you run the script, can you show me the output of

ray status
ray list nodes
ray list actors

?

Can you also try restarting ray?

ray stop
ray start

[iamjustinhsu]

oh wait, you're machine has 16GiB 12CPU, and mine has 24GiB, let me see if I can mock that out and reproduce.

[HenryL27]

❯ poetry run ray status
======== Autoscaler status: 2025-10-22 11:53:32.999004 ========
Node status
---------------------------------------------------------------
Active:
 1 node_e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317
Pending:
 (no pending nodes)
Recent failures:
 (no failures)

Resources
---------------------------------------------------------------
Total Usage:
 1.75/10.0 CPU
 7.00GiB/7.82GiB memory
 476.87KiB/2.00GiB object_store_memory

From request_resources:
 {'CPU': 1}: 16 from request_resources()
Pending Demands:
 {'CPU': 0.25, 'memory': 1073741824.0}: 3+ pending tasks/actors

❯ poetry run ray list nodes

======== List: 2025-10-22 11:54:04.551308 ========
Stats:
------------------------------
Total: 1

Table:
------------------------------
    NODE_ID                                                   NODE_IP    IS_HEAD_NODE    STATE    STATE_MESSAGE    NODE_NAME    RESOURCES_TOTAL                 LABELS
 0  e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  127.0.0.1  True            ALIVE                     127.0.0.1    CPU: 10.0                       ray.io/node-id: e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317
                                                                                                                                memory: 7.815 GiB
                                                                                                                                node:127.0.0.1: 1.0
                                                                                                                                node:__internal_head__: 1.0
                                                                                                                                object_store_memory: 2.000 GiB

❯ poetry run ray list actors

======== List: 2025-10-22 11:54:29.727249 ========
Stats:
------------------------------
Total: 12

Table:
------------------------------
    ACTOR_ID                          CLASS_NAME             STATE               JOB_ID  NAME                  NODE_ID                                                     PID  RAY_NAMESPACE
 0  1820ee1bd200cfa46f80168701000000  HashShuffleAggregator  PENDING_CREATION  01000000                                                                                      0  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 1  1afe343d074f26044ec81c8301000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55945  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 2  1e10daf0919e5ae9d696076d01000000  AutoscalingRequester   ALIVE             01000000  AutoscalingRequester  e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55946  AutoscalingRequester
 3  4041b6a0bef268e9d6fbd40801000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55944  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 4  41c2b520bfc2cbc6653752e001000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55952  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 5  650a0b35140a403d6f1e156101000000  HashShuffleAggregator  PENDING_CREATION  01000000                                                                                      0  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 6  701169f205606cbb63984e4401000000  _StatsActor            ALIVE             01000000  datasets_stats_actor  e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55950  _dataset_stats_actor
 7  831090a16c475807cdc9908f01000000  HashShuffleAggregator  PENDING_CREATION  01000000                                                                                      0  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 8  928eaa25d9ecdcb4ae78ed9001000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55953  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
 9  a5021df885c247afa08ea1c801000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55947  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
10  cd54c6ad292b66ec4b4e9c1301000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55949  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c
11  e7b31c96541f5f34e26519b201000000  HashShuffleAggregator  ALIVE             01000000                        e33ab5e33cde362c62d2c883cd078add7cc33d89fb76aad48629c317  55948  1bc0c8f7-0d18-4e1f-9feb-4544de34e81c