presto-main/src/main/java/io/prestosql/operator/WorkProcessorPipelineSourceOperator.java/WorkProcessorPipelineSourceOperator/workProcessorOperatorStateMonitor
class WorkProcessorPipelineSourceOperator:    private void workProcessorOperatorStateMonitor(WorkProcessor.ProcessState<Page> state, int operatorIndex)
    {
        // report timing for current operator
        WorkProcessorOperatorContext context = workProcessorOperatorContexts.get(operatorIndex);
        timer.recordOperationComplete(context.operatorTiming);

        if (operatorIndex == 0) {
            // update input stats for source operator
            WorkProcessorSourceOperator sourceOperator = (WorkProcessorSourceOperator) context.operator;

            long deltaPhysicalInputDataSize = deltaAndSet(context.physicalInputDataSize, sourceOperator.getPhysicalInputDataSize().toBytes());
            long deltaPhysicalInputPositions = deltaAndSet(context.physicalInputPositions, sourceOperator.getPhysicalInputPositions());

            long deltaInternalNetworkInputDataSize = deltaAndSet(context.internalNetworkInputDataSize, sourceOperator.getInternalNetworkInputDataSize().toBytes());
            long deltaInternalNetworkInputPositions = deltaAndSet(context.internalNetworkInputPositions, sourceOperator.getInternalNetworkPositions());

            long deltaInputDataSize = deltaAndSet(context.inputDataSize, sourceOperator.getInputDataSize().toBytes());
            long deltaInputPositions = deltaAndSet(context.inputPositions, sourceOperator.getInputPositions());

            long deltaReadTimeNanos = deltaAndSet(context.readTimeNanos, sourceOperator.getReadTime().roundTo(NANOSECONDS));

            operatorContext.recordPhysicalInputWithTiming(deltaPhysicalInputDataSize, deltaPhysicalInputPositions, deltaReadTimeNanos);
            operatorContext.recordNetworkInput(deltaInternalNetworkInputDataSize, deltaInternalNetworkInputPositions);
            operatorContext.recordProcessedInput(deltaInputDataSize, deltaInputPositions);
        }

        if (state.getType() == FINISHED) {
            // immediately close all upstream operators (including finished operator)
            closeOperators(operatorIndex);
        }
        else if (state.getType() == BLOCKED && blockedFuture != state.getBlocked()) {
            blockedFuture = state.getBlocked();
            long start = System.nanoTime();
            blockedFuture.addListener(
                    () -> context.blockedWallNanos.getAndAdd(System.nanoTime() - start),
                    directExecutor());
        }
    }