presto-main/src/main/java/io/prestosql/operator/WorkProcessorPipelineSourceOperator.java/WorkProcessorPipelineSourceOperator/getNestedOperatorStats
class WorkProcessorPipelineSourceOperator:    private List<OperatorStats> getNestedOperatorStats()
    {
        return workProcessorOperatorContexts.stream()
                .map(context -> new OperatorStats(
                        stageId,
                        pipelineId,
                        context.operatorId,
                        context.planNodeId,
                        context.operatorType,
                        1,

                        // WorkProcessorOperator doesn't have addInput call
                        0,
                        // source operators report read time though
                        new Duration(context.readTimeNanos.get(), NANOSECONDS),
                        ZERO_DURATION,

                        succinctBytes(context.physicalInputDataSize.get()),
                        context.physicalInputPositions.get(),

                        succinctBytes(context.internalNetworkInputDataSize.get()),
                        context.internalNetworkInputPositions.get(),

                        succinctBytes(context.physicalInputDataSize.get() + context.internalNetworkInputDataSize.get()),

                        succinctBytes(context.inputDataSize.get()),
                        context.inputPositions.get(),
                        context.inputPositions.get() * context.inputPositions.get(),

                        context.operatorTiming.getCalls(),
                        new Duration(context.operatorTiming.getWallNanos(), NANOSECONDS),
                        new Duration(context.operatorTiming.getCpuNanos(), NANOSECONDS),

                        succinctBytes(context.outputDataSize.get()),
                        context.outputPositions.get(),

                        new DataSize(0, BYTE),

                        new Duration(context.blockedWallNanos.get(), NANOSECONDS),

                        // WorkProcessorOperator doesn't have finish call
                        0,
                        ZERO_DURATION,
                        ZERO_DURATION,

                        succinctBytes(context.memoryTrackingContext.getUserMemory()),
                        succinctBytes(context.memoryTrackingContext.getRevocableMemory()),
                        succinctBytes(context.memoryTrackingContext.getSystemMemory()),
                        succinctBytes(context.peakUserMemoryReservation.get()),
                        succinctBytes(context.peakSystemMemoryReservation.get()),
                        succinctBytes(context.peakRevocableMemoryReservation.get()),
                        succinctBytes(context.peakTotalMemoryReservation.get()),
                        new DataSize(0, BYTE),
                        operatorContext.isWaitingForMemory().isDone() ? Optional.empty() : Optional.of(WAITING_FOR_MEMORY),
                        getOperatorInfo(context)))
                .collect(toImmutableList());
    }