presto-main/src/main/java/io/prestosql/operator/WorkProcessorPipelineSourceOperator.java/WorkProcessorPipelineSourceOperator/recordProcessedOutput
class WorkProcessorPipelineSourceOperator:    private Page recordProcessedOutput(Page page, int operatorIndex)
    {
        WorkProcessorOperatorContext operatorContext = workProcessorOperatorContexts.get(operatorIndex);
        operatorContext.outputPositions.getAndAdd(page.getPositionCount());

        WorkProcessorOperatorContext downstreamOperatorContext;
        if (!isLastOperator(operatorIndex)) {
            downstreamOperatorContext = workProcessorOperatorContexts.get(operatorIndex + 1);
            downstreamOperatorContext.inputPositions.getAndAdd(page.getPositionCount());
        }
        else {
            downstreamOperatorContext = null;
        }

        // account processed bytes from lazy blocks only when they are loaded
        recordMaterializedBytes(page, sizeInBytes -> {
            operatorContext.outputDataSize.getAndAdd(sizeInBytes);
            if (downstreamOperatorContext != null) {
                downstreamOperatorContext.inputDataSize.getAndAdd(sizeInBytes);
            }
        });

        return page;
    }