presto-main/src/main/java/io/prestosql/operator/WorkProcessorUtils.java/WorkProcessorUtils/transform
class WorkProcessorUtils:    static <T, R> WorkProcessor<R> transform(WorkProcessor<T> processor, Transformation<T, R> transformation)
    {
        requireNonNull(processor, "processor is null");
        requireNonNull(transformation, "transformation is null");
        return create(new WorkProcessor.Process<R>()
        {
            T element;

            @Override
            public ProcessState<R> process()
            {
                while (true) {
                    if (element == null && !processor.isFinished()) {
                        if (processor.process()) {
                            if (!processor.isFinished()) {
                                element = requireNonNull(processor.getResult(), "result is null");
                            }
                        }
                        else if (processor.isBlocked()) {
                            return ProcessState.blocked(processor.getBlockedFuture());
                        }
                        else {
                            return ProcessState.yield();
                        }
                    }

                    TransformationState<R> state = requireNonNull(transformation.process(element), "state is null");

                    if (state.isNeedsMoreData()) {
                        checkState(!processor.isFinished(), "Cannot request more data when base processor is finished");
                        // set element to empty() in order to fetch a new one
                        element = null;
                    }

                    // pass-through transformation state if it doesn't require new data
                    switch (state.getType()) {
                        case NEEDS_MORE_DATA:
                            break;
                        case BLOCKED:
                            return ProcessState.blocked(state.getBlocked());
                        case YIELD:
                            return ProcessState.yield();
                        case RESULT:
                            return ProcessState.ofResult(state.getResult());
                        case FINISHED:
                            return ProcessState.finished();
                    }
                }
            }
        });
    }