presto-main/src/main/java/io/prestosql/execution/scheduler/UniformNodeSelector.java/UniformNodeSelector/equateDistribution
class UniformNodeSelector:    private void equateDistribution(Multimap<InternalNode, Split> assignment, NodeAssignmentStats assignmentStats, NodeMap nodeMap)
    {
        if (assignment.isEmpty()) {
            return;
        }

        Collection<InternalNode> allNodes = nodeMap.getNodesByHostAndPort().values();
        if (allNodes.size() < 2) {
            return;
        }

        IndexedPriorityQueue<InternalNode> maxNodes = new IndexedPriorityQueue<>();
        for (InternalNode node : assignment.keySet()) {
            maxNodes.addOrUpdate(node, assignmentStats.getTotalSplitCount(node));
        }

        IndexedPriorityQueue<InternalNode> minNodes = new IndexedPriorityQueue<>();
        for (InternalNode node : allNodes) {
            minNodes.addOrUpdate(node, Long.MAX_VALUE - assignmentStats.getTotalSplitCount(node));
        }

        while (true) {
            if (maxNodes.isEmpty()) {
                return;
            }

            // fetch min and max node
            InternalNode maxNode = maxNodes.poll();
            InternalNode minNode = minNodes.poll();

            if (assignmentStats.getTotalSplitCount(maxNode) - assignmentStats.getTotalSplitCount(minNode) <= 1) {
                return;
            }

            // move split from max to min
            redistributeSplit(assignment, maxNode, minNode, nodeMap.getNodesByHost());
            assignmentStats.removeAssignedSplit(maxNode);
            assignmentStats.addAssignedSplit(minNode);

            // add max back into maxNodes only if it still has assignments
            if (assignment.containsKey(maxNode)) {
                maxNodes.addOrUpdate(maxNode, assignmentStats.getTotalSplitCount(maxNode));
            }

            // Add or update both the Priority Queues with the updated node priorities
            maxNodes.addOrUpdate(minNode, assignmentStats.getTotalSplitCount(minNode));
            minNodes.addOrUpdate(minNode, Long.MAX_VALUE - assignmentStats.getTotalSplitCount(minNode));
            minNodes.addOrUpdate(maxNode, Long.MAX_VALUE - assignmentStats.getTotalSplitCount(maxNode));
        }
    }