[Data] Sample finalized partitions randomly to avoid lensing finalization on a single node

[alexeykudinkin]

Thank you for contributing to Ray! 🚀
Please review the Ray Contribution Guide before opening a pull request.

⚠️ Remove these instructions before submitting your PR.

💡 Tip: Mark as draft if you want early feedback, or ready for review when it's complete.

Description

Currently, finalization is scheduled in batches sequentially -- ie batch of N adjacent partitions is finalized at once (in a sliding window).

This creates a lensing effect since:

1. Adjacent partitions i and i+1 get scheduled onto adjacent aggregators j and j+i (since membership is determined as j = i % num_aggregators)
2. Adjacent aggregators have high likelihood of getting scheduled on the same node (due to similarly being scheduled at about the same time in sequence)

To address that this change applies random sampling when choosing next partitions to finalize to make sure partitions are chosen uniformly reducing concurrent finalization of the adjacent partitions.

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request refactors the finalization logic in the hash shuffle operator to randomly sample partitions instead of processing them sequentially. This is a valuable change that should help distribute the finalization load more evenly across the cluster and avoid potential node hotspots. My review includes one suggestion to optimize the new sampling logic for better performance.

[gemini-code-assist]

The random.sample function can operate directly on sets, so converting self._pending_finalization_partition_ids to a list is unnecessary. Removing the list() conversion will improve performance by avoiding the creation of a new list in each call, which can be expensive if the number of pending partitions is large.

Suggested change

	target_partition_ids = random.sample(
	list(self._pending_finalization_partition_ids), next_batch_size
	)
	target_partition_ids = random.sample(
	self._pending_finalization_partition_ids, next_batch_size
	)

[iamjustinhsu]

wait is this true? if module N = num actors, then partition i and i + 1 must necessarily be in different actors. Oh wait nvm, i see what your saying

[iamjustinhsu]

is this necessarily true? your default strategy is spread, and each aggregator is scheduled with same num of resources, so aggregator i and i + 1 have as much of a chance of scheduling on the same node as aggregator i and j. please correct my assumptions if im wrong

[goutamvenkat-anyscale]

So wouldn't a better strategy be to check how much each agg actor is currently consuming relative to the node's capacity and schedule the finalization if there's remaining capacity?

I just find the randomization strategy harder to reason in this case.

[goutamvenkat-anyscale]

Also it's a function of partition size, so ideally if we can get metadata about the partition before scheduling the finalize() that would be even better.