presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSinkOperator.java/LocalExchangeSinkOperator/addInput
class LocalExchangeSinkOperator:    @Override
    public void addInput(Page page)
    {
        requireNonNull(page, "page is null");
        page = pagePreprocessor.apply(page);
        sink.addPage(page);
        operatorContext.recordOutput(page.getSizeInBytes(), page.getPositionCount());
    }