presto-main/src/main/java/io/prestosql/operator/project/MergePages.java/MergePagesTransformation/process
class MergePagesTransformation:        @Override
        public TransformationState<Page> process(Page inputPage)
        {
            if (queuedPage != null) {
                Page output = queuedPage;
                queuedPage = null;
                memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
                return ofResult(output);
            }

            boolean inputFinished = inputPage == null;
            if (inputFinished) {
                if (pageBuilder.isEmpty()) {
                    memoryContext.close();
                    return finished();
                }

                return ofResult(flush(), false);
            }

            // TODO: merge low cardinality blocks lazily
            if (inputPage.getPositionCount() >= minRowCount || !isLoaded(inputPage) || inputPage.getSizeInBytes() >= minPageSizeInBytes) {
                if (pageBuilder.isEmpty()) {
                    return ofResult(inputPage);
                }

                Page output = pageBuilder.build();
                pageBuilder.reset();
                // inputPage is preserved until next process(...) call
                queuedPage = inputPage;
                memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes() + inputPage.getRetainedSizeInBytes());
                return ofResult(output, false);
            }

            appendPage(inputPage);

            if (pageBuilder.isFull()) {
                return ofResult(flush());
            }

            memoryContext.setBytes(pageBuilder.getRetainedSizeInBytes());
            return needsMoreData();
        }