presto-main/src/main/java/io/prestosql/operator/FilterAndProjectOperator.java/FilterAndProjectOperator/FilterAndProjectOperator
class FilterAndProjectOperator:    public FilterAndProjectOperator(
            Session session,
            MemoryTrackingContext memoryTrackingContext,
            DriverYieldSignal yieldSignal,
            Optional<WorkProcessor<Page>> sourcePages,
            PageProcessor pageProcessor,
            List<Type> types,
            DataSize minOutputPageSize,
            int minOutputPageRowCount)
    {
        AggregatedMemoryContext localAggregatedMemoryContext = newSimpleAggregatedMemoryContext();
        LocalMemoryContext outputMemoryContext = localAggregatedMemoryContext.newLocalMemoryContext(FilterAndProjectOperator.class.getSimpleName());

        this.pages = sourcePages.orElse(pageBuffer.pages())
                .flatMap(page -> pageProcessor.createWorkProcessor(
                        session.toConnectorSession(),
                        yieldSignal,
                        outputMemoryContext,
                        page,
                        sourcePages.isPresent()))
                .transformProcessor(processor -> mergePages(types, minOutputPageSize.toBytes(), minOutputPageRowCount, processor, localAggregatedMemoryContext))
                .withProcessStateMonitor(state -> memoryTrackingContext.localSystemMemoryContext().setBytes(localAggregatedMemoryContext.getBytes()));
    }