presto-main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushProjectionIntoTableScan.java/PushProjectionIntoTableScan/apply
class PushProjectionIntoTableScan:    @Override
    public Result apply(ProjectNode project, Captures captures, Context context)
    {
        TableScanNode tableScan = captures.get(TABLE_SCAN);

        Map<Symbol, Expression> inputExpressions = project.getAssignments().getMap();

        ImmutableList.Builder<NodeRef<Expression>> nodeReferencesBuilder = ImmutableList.builder();
        ImmutableList.Builder<ConnectorExpression> partialProjectionsBuilder = ImmutableList.builder();

        // Extract translatable components from projection expressions. Prepare a mapping from these internal
        // expression nodes to corresponding ConnectorExpression translations.
        for (Map.Entry<Symbol, Expression> expression : inputExpressions.entrySet()) {
            Map<NodeRef<Expression>, ConnectorExpression> partialTranslations = extractPartialTranslations(
                    expression.getValue(),
                    context.getSession(),
                    typeAnalyzer,
                    context.getSymbolAllocator().getTypes());

            partialTranslations.forEach((nodeRef, expr) -> {
                nodeReferencesBuilder.add(nodeRef);
                partialProjectionsBuilder.add(expr);
            });
        }

        List<NodeRef<Expression>> nodesForPartialProjections = nodeReferencesBuilder.build();
        List<ConnectorExpression> connectorPartialProjections = partialProjectionsBuilder.build();

        Map<String, ColumnHandle> assignments = tableScan.getAssignments()
                .entrySet().stream()
                .collect(toImmutableMap(entry -> entry.getKey().getName(), Map.Entry::getValue));

        Optional<ProjectionApplicationResult<TableHandle>> result = metadata.applyProjection(context.getSession(), tableScan.getTable(), connectorPartialProjections, assignments);

        if (!result.isPresent()) {
            return Result.empty();
        }

        List<ConnectorExpression> newConnectorPartialProjections = result.get().getProjections();
        checkState(newConnectorPartialProjections.size() == connectorPartialProjections.size(),
                "Mismatch between input and output projections from the connector: expected %s but got %s",
                connectorPartialProjections.size(),
                newConnectorPartialProjections.size());

        List<Symbol> newScanOutputs = new ArrayList<>();
        Map<Symbol, ColumnHandle> newScanAssignments = new HashMap<>();
        Map<String, Symbol> variableMappings = new HashMap<>();
        for (ProjectionApplicationResult.Assignment assignment : result.get().getAssignments()) {
            Symbol symbol = context.getSymbolAllocator().newSymbol(assignment.getVariable(), assignment.getType());

            newScanOutputs.add(symbol);
            newScanAssignments.put(symbol, assignment.getColumn());
            variableMappings.put(assignment.getVariable(), symbol);
        }

        // Translate partial connector projections back to new partial projections
        List<Expression> newPartialProjections = newConnectorPartialProjections.stream()
                .map(expression -> ConnectorExpressionTranslator.translate(expression, variableMappings, new LiteralEncoder(metadata)))
                .collect(toImmutableList());

        // Map internal node references to new partial projections
        ImmutableMap.Builder<NodeRef<Expression>, Expression> nodesToNewPartialProjectionsBuilder = ImmutableMap.builder();
        for (int i = 0; i < nodesForPartialProjections.size(); i++) {
            nodesToNewPartialProjectionsBuilder.put(nodesForPartialProjections.get(i), newPartialProjections.get(i));
        }
        Map<NodeRef<Expression>, Expression> nodesToNewPartialProjections = nodesToNewPartialProjectionsBuilder.build();

        // Stitch partial translations to form new complete projections
        Assignments.Builder newProjectionAssignments = Assignments.builder();
        project.getAssignments().entrySet().forEach(entry -> {
            newProjectionAssignments.put(entry.getKey(), replaceExpression(entry.getValue(), nodesToNewPartialProjections));
        });

        return Result.ofPlanNode(
                new ProjectNode(
                        context.getIdAllocator().getNextId(),
                        TableScanNode.newInstance(
                                tableScan.getId(),
                                result.get().getHandle(),
                                newScanOutputs,
                                newScanAssignments),
                        newProjectionAssignments.build()));
    }