presto-main/src/main/java/io/prestosql/util/MergeSortedPages.java/MergeSortedPages/buildPage
class MergeSortedPages:    private static WorkProcessor<Page> buildPage(
            WorkProcessor<PageWithPosition> pageWithPositions,
            List<Integer> outputChannels,
            List<Type> outputTypes,
            BiPredicate<PageBuilder, PageWithPosition> pageBreakPredicate,
            boolean updateMemoryAfterEveryPosition,
            AggregatedMemoryContext aggregatedMemoryContext,
            DriverYieldSignal yieldSignal)
    {
        LocalMemoryContext memoryContext = aggregatedMemoryContext.newLocalMemoryContext(MergeSortedPages.class.getSimpleName());
        PageBuilder pageBuilder = new PageBuilder(outputTypes);
        return pageWithPositions
                .yielding(yieldSignal::isSet)
                .transform(pageWithPosition -> {
                    boolean finished = pageWithPosition == null;
                    if (finished && pageBuilder.isEmpty()) {
                        memoryContext.close();
                        return TransformationState.finished();
                    }

                    if (finished || pageBreakPredicate.test(pageBuilder, pageWithPosition)) {
                        if (!updateMemoryAfterEveryPosition) {
                            // update memory usage just before producing page to cap from top
                            memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
                        }

                        Page page = pageBuilder.build();
                        pageBuilder.reset();
                        if (!finished) {
                            pageWithPosition.appendTo(pageBuilder, outputChannels, outputTypes);
                        }

                        if (updateMemoryAfterEveryPosition) {
                            memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
                        }

                        return TransformationState.ofResult(page, !finished);
                    }

                    pageWithPosition.appendTo(pageBuilder, outputChannels, outputTypes);

                    if (updateMemoryAfterEveryPosition) {
                        memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
                    }

                    return TransformationState.needsMoreData();
                });
    }