presto-main/src/main/java/io/prestosql/operator/WorkProcessorUtils.java/WorkProcessorUtils/mergeSorted
class WorkProcessorUtils:    static <T> WorkProcessor<T> mergeSorted(Iterable<WorkProcessor<T>> processorIterable, Comparator<T> comparator)
    {
        requireNonNull(comparator, "comparator is null");
        Iterator<WorkProcessor<T>> processorIterator = requireNonNull(processorIterable, "processorIterable is null").iterator();
        checkArgument(processorIterator.hasNext(), "There must be at least one base processor");
        PriorityQueue<ElementAndProcessor<T>> queue = new PriorityQueue<>(2, comparing(ElementAndProcessor::getElement, comparator));

        return create(new WorkProcessor.Process<T>()
        {
            WorkProcessor<T> processor = requireNonNull(processorIterator.next());

            @Override
            public ProcessState<T> process()
            {
                while (true) {
                    if (processor.process()) {
                        if (!processor.isFinished()) {
                            queue.add(new ElementAndProcessor<>(processor.getResult(), processor));
                        }
                    }
                    else if (processor.isBlocked()) {
                        return ProcessState.blocked(processor.getBlockedFuture());
                    }
                    else {
                        return ProcessState.yield();
                    }

                    if (processorIterator.hasNext()) {
                        processor = requireNonNull(processorIterator.next());
                        continue;
                    }

                    if (queue.isEmpty()) {
                        return ProcessState.finished();
                    }

                    ElementAndProcessor<T> elementAndProcessor = queue.poll();
                    processor = elementAndProcessor.getProcessor();
                    return ProcessState.ofResult(elementAndProcessor.getElement());
                }
            }
        });
    }