Hi team, I’m new here. I’m trying to use Daft on Ray to do table joins. Even though I explicitly have no worker on the head node (--num-cpus=0), and that I do observe that all the ScanWithTask, Project, FanoutHash, etc are being executed on other workers, for some reason, the head node memory keeps increasing and eventually goes OOM and takes the whole cluster down with it. I’m trying to do a left join. Any insights would be so much appreciated! Thanks!

Hey @Henry T! How many partitions are you running this join on? Ray does use quite a bit of memory (on the head node) when running joins on large numbers of partitions.

According to the physical plan:

FanoutByHash: 64065

Yup this is likely choking the Ray head node because of the sheer number of Ray objects. Could you also provide information about the input number of partitions (the plan before the Fanout?). @Sammy Sidhu is working on a new shuffle mechanism (draft PR that already has some really promising results) that will help us work around this much better for shuffles with a large number of partitions.

Yup, this is the physical plans:

== Physical Plan ==

* HashJoin: Type = Left
|   On = col(image_url)
|\
| * ReduceMerge
| |
| * FanoutByHash: 64065
| |   Partition by = col(image_url)
| |
| * Project: ...
| |   Clustering spec = { Num partitions = 64 }
| |
| * TabularScan:
| |   Num Scan Tasks = 64
| |   Estimated Scan Bytes = 11987515627
| |   Clustering spec = { Num partitions = 64 }
| |   Schema: {...}
| |   Scan Tasks: [...]
|
* ReduceMerge
|
* FanoutByHash: 64065
|   Partition by = col(image_url)
|
* Explode: col(image_url)
|   Clustering spec = { Num partitions = 64065 }
|
* Project: ...
|   Clustering spec = { Num partitions = 64065 }
|
* TabularScan:
|   Num Scan Tasks = 64065
|   Estimated Scan Bytes = 3253337041356
|   Clustering spec = { Num partitions = 64065 }
|   Schema: {...}
|   Scan Tasks: [...]

Thanks this is super helpful! The Ray head node doesn’t store any intermediate data, but surprisingly Ray ObjectRefs themselves are pretty heavyweight (I think in the order of 3KB per object) so just storing these “pointers” in a Ray object graph can get quite large. So it’s actually OOMing on that. In this case it would seem that there are just too many intermediate objects being generated on Ray which is struggling to keep up. The new algorithm I described earlier should alleviate the problem significantly. Would love to have you try it out when we have it (soon!). I’ll keep you updated

that would be awesome thank you

@jay do you have a sense of how many partitions are currently supported for a join on daft + ray for a given memory configuration?

1,000 x 1,000 is usually pretty reasonable on something like a 32GB head node!