class PruneApplySourceColumns implements Rule<ApplyNode> {

    @Override
    public Result apply(ApplyNode applyNode, Captures captures, Context context)
    {
        Set<Symbol> subqueryAssignmentsSymbols = applyNode.getSubqueryAssignments().getExpressions().stream()
                .flatMap(expression -> SymbolsExtractor.extractUnique(expression).stream())
                .collect(toImmutableSet());

        Optional<PlanNode> prunedSubquery = restrictOutputs(context.getIdAllocator(), applyNode.getSubquery(), subqueryAssignmentsSymbols);
        return prunedSubquery.map(subquery -> applyNode.replaceChildren(ImmutableList.of(applyNode.getInput(), subquery)))
                .map(Result::ofPlanNode)
                .orElse(Result.empty());
    }

}