presto-main/src/main/java/io/prestosql/operator/exchange/PartitioningExchanger.java/PartitioningExchanger/accept
class PartitioningExchanger:    @Override
    public synchronized void accept(Page page)
    {
        // reset the assignment lists
        for (IntList partitionAssignment : partitionAssignments) {
            partitionAssignment.clear();
        }

        // assign each row to a partition
        for (int position = 0; position < page.getPositionCount(); position++) {
            int partition = partitionGenerator.getPartition(page, position);
            partitionAssignments[partition].add(position);
        }

        // build a page for each partition
        Block[] outputBlocks = new Block[page.getChannelCount()];
        for (int partition = 0; partition < buffers.size(); partition++) {
            IntArrayList positions = partitionAssignments[partition];
            if (!positions.isEmpty()) {
                for (int i = 0; i < page.getChannelCount(); i++) {
                    outputBlocks[i] = page.getBlock(i).copyPositions(positions.elements(), 0, positions.size());
                }

                Page pageSplit = new Page(positions.size(), outputBlocks);
                memoryManager.updateMemoryUsage(pageSplit.getRetainedSizeInBytes());
                buffers.get(partition).accept(new PageReference(pageSplit, 1, () -> memoryManager.updateMemoryUsage(-pageSplit.getRetainedSizeInBytes())));
            }
        }
    }