presto-main/src/main/java/io/prestosql/operator/exchange/LocalMergeSourceOperator.java/LocalMergeSourceOperatorFactory/createOperator
class LocalMergeSourceOperatorFactory:        @Override
        public Operator createOperator(DriverContext driverContext)
        {
            checkState(!closed, "Factory is already closed");

            LocalExchange localExchange = localExchangeFactory.getLocalExchange(driverContext.getLifespan());

            OperatorContext operatorContext = driverContext.addOperatorContext(operatorId, planNodeId, LocalMergeSourceOperator.class.getSimpleName());
            PageWithPositionComparator comparator = orderingCompiler.compilePageWithPositionComparator(types, sortChannels, orderings);
            List<LocalExchangeSource> sources = IntStream.range(0, localExchange.getBufferCount())
                    .boxed()
                    .map(index -> localExchange.getNextSource())
                    .collect(toImmutableList());
            return new LocalMergeSourceOperator(operatorContext, sources, types, comparator);
        }