presto-main/src/main/java/io/prestosql/operator/TaskOutputOperator.java/TaskOutputOperator/addInput
class TaskOutputOperator:    @Override
    public void addInput(Page page)
    {
        requireNonNull(page, "page is null");
        if (page.getPositionCount() == 0) {
            return;
        }

        page = pagePreprocessor.apply(page);

        List<SerializedPage> serializedPages = splitPage(page, DEFAULT_MAX_PAGE_SIZE_IN_BYTES).stream()
                .map(serde::serialize)
                .collect(toImmutableList());

        outputBuffer.enqueue(serializedPages);
        operatorContext.recordOutput(page.getSizeInBytes(), page.getPositionCount());
    }