presto-main/src/main/java/io/prestosql/operator/WorkProcessorUtils.java/WorkProcessorUtils/flatten
class WorkProcessorUtils:    static <T> WorkProcessor<T> flatten(WorkProcessor<WorkProcessor<T>> processor)
    {
        requireNonNull(processor, "processor is null");
        return processor.transform(nestedProcessor -> {
            if (nestedProcessor == null) {
                return TransformationState.finished();
            }

            if (nestedProcessor.process()) {
                if (nestedProcessor.isFinished()) {
                    return TransformationState.needsMoreData();
                }

                return TransformationState.ofResult(nestedProcessor.getResult(), false);
            }

            if (nestedProcessor.isBlocked()) {
                return TransformationState.blocked(nestedProcessor.getBlockedFuture());
            }

            return TransformationState.yield();
        });
    }