class WorkProcessorPipelineSourceOperator implements SourceOperator {

    private WorkProcessorPipelineSourceOperator(
            int operatorId,
            DriverContext driverContext,
            WorkProcessorSourceOperatorFactory sourceOperatorFactory,
            List<WorkProcessorOperatorFactory> operatorFactories)
    {
        requireNonNull(driverContext, "driverContext is null");
        requireNonNull(sourceOperatorFactory, "sourceOperatorFactory is null");
        requireNonNull(operatorFactories, "operatorFactories is null");
        this.stageId = driverContext.getTaskId().getStageId().getId();
        this.pipelineId = driverContext.getPipelineContext().getPipelineId();
        this.sourceId = requireNonNull(sourceOperatorFactory.getSourceId(), "sourceId is null");
        this.operatorContext = driverContext.addOperatorContext(operatorId, sourceId, WorkProcessorPipelineSourceOperator.class.getSimpleName());
        this.timer = new OperationTimer(
                operatorContext.getDriverContext().isCpuTimerEnabled(),
                operatorContext.getDriverContext().isCpuTimerEnabled() && operatorContext.getDriverContext().isPerOperatorCpuTimerEnabled());

        MemoryTrackingContext sourceOperatorMemoryTrackingContext = createMemoryTrackingContext(operatorContext, 0);
        sourceOperatorMemoryTrackingContext.initializeLocalMemoryContexts(sourceOperatorFactory.getOperatorType());
        WorkProcessor<Split> splits = WorkProcessor.create(new Splits());

        sourceOperator = sourceOperatorFactory.create(
                operatorContext.getSession(),
                sourceOperatorMemoryTrackingContext,
                operatorContext.getDriverContext().getYieldSignal(),
                splits);
        workProcessorOperatorContexts.add(new WorkProcessorOperatorContext(
                sourceOperator,
                sourceOperatorFactory.getOperatorId(),
                sourceOperatorFactory.getPlanNodeId(),
                sourceOperatorFactory.getOperatorType(),
                sourceOperatorMemoryTrackingContext));
        WorkProcessor<Page> pages = sourceOperator.getOutputPages();
        pages = pages
                .yielding(() -> operatorContext.getDriverContext().getYieldSignal().isSet())
                .withProcessEntryMonitor(() -> workProcessorOperatorEntryMonitor(0))
                .withProcessStateMonitor(state -> workProcessorOperatorStateMonitor(state, 0))
                .map(page -> recordProcessedOutput(page, 0));

        for (int i = 0; i < operatorFactories.size(); ++i) {
            int operatorIndex = i + 1;
            WorkProcessorOperatorFactory operatorFactory = operatorFactories.get(i);
            MemoryTrackingContext operatorMemoryTrackingContext = createMemoryTrackingContext(operatorContext, operatorIndex);
            operatorMemoryTrackingContext.initializeLocalMemoryContexts(operatorFactory.getOperatorType());
            WorkProcessorOperator operator = operatorFactory.create(
                    new ProcessorContext(operatorContext.getSession(), operatorMemoryTrackingContext, operatorContext),
                    pages);
            workProcessorOperatorContexts.add(new WorkProcessorOperatorContext(
                    operator,
                    operatorFactory.getOperatorId(),
                    operatorFactory.getPlanNodeId(),
                    operatorFactory.getOperatorType(),
                    operatorMemoryTrackingContext));
            pages = operator.getOutputPages();
            pages = pages
                    .yielding(() -> operatorContext.getDriverContext().getYieldSignal().isSet())
                    .withProcessEntryMonitor(() -> workProcessorOperatorEntryMonitor(operatorIndex))
                    .withProcessStateMonitor(state -> workProcessorOperatorStateMonitor(state, operatorIndex));
            pages = pages.map(page -> recordProcessedOutput(page, operatorIndex));
        }

        // materialize output pages as there are no semantics guarantees for non WorkProcessor operators
        pages = pages.map(Page::getLoadedPage);

        // finish early when entire pipeline is closed
        this.pages = pages.finishWhen(() -> operatorFinishing);

        operatorContext.setNestedOperatorStatsSupplier(this::getNestedOperatorStats);
    }

    private  WorkProcessorPipelineSourceOperator(
            int operatorId,
            DriverContext driverContext,
            WorkProcessorSourceOperatorFactory sourceOperatorFactory,
            List<WorkProcessorOperatorFactory> operatorFactories);

}