paimon-presto-0.236/src/main/java/org/apache/paimon/presto/PrestoComputePushdown.java [54:180]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class PrestoComputePushdown implements ConnectorPlanOptimizer {

    private final StandardFunctionResolution functionResolution;
    private final RowExpressionService rowExpressionService;
    private final FunctionMetadataManager functionMetadataManager;
    private final PrestoTransactionManager transactionManager;

    public PrestoComputePushdown(
            StandardFunctionResolution functionResolution,
            RowExpressionService rowExpressionService,
            FunctionMetadataManager functionMetadataManager,
            PrestoTransactionManager transactionManager) {

        this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
        this.rowExpressionService =
                requireNonNull(rowExpressionService, "rowExpressionService is null");
        this.functionMetadataManager =
                requireNonNull(functionMetadataManager, "functionMetadataManager is null");
        this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
    }

    @Override
    public PlanNode optimize(
            PlanNode maxSubplan,
            ConnectorSession session,
            VariableAllocator variableAllocator,
            PlanNodeIdAllocator idAllocator) {

        if (isPaimonPushdownEnabled(session)) {
            return maxSubplan.accept(new Visitor(session, idAllocator), null);
        }
        return maxSubplan;
    }

    private class Visitor extends PlanVisitor<PlanNode, Void> {

        private final ConnectorSession session;
        private final PlanNodeIdAllocator idAllocator;

        public Visitor(ConnectorSession session, PlanNodeIdAllocator idAllocator) {
            this.session = requireNonNull(session, "session is null");
            this.idAllocator = requireNonNull(idAllocator, "idAllocator is null");
        }

        private Optional<List<ColumnHandle>> extractColumns(
                RowExpression expression,
                Map<VariableReferenceExpression, ColumnHandle> assignments) {

            if (expression instanceof VariableReferenceExpression) {
                VariableReferenceExpression variable = (VariableReferenceExpression) expression;
                ColumnHandle columnHandle = assignments.get(variable);
                if (columnHandle != null) {
                    return Optional.of(Collections.singletonList(columnHandle));
                }
                return Optional.empty();
            } else if (expression instanceof CallExpression) {
                CallExpression callExpression = (CallExpression) expression;
                List<ColumnHandle> columns = new ArrayList<>();
                for (RowExpression argument : callExpression.getArguments()) {
                    extractColumns(argument, assignments).ifPresent(columns::addAll);
                }
                return Optional.of(columns);
            }
            return Optional.empty();
        }

        @Override
        public PlanNode visitPlan(PlanNode node, Void context) {
            ImmutableList.Builder<PlanNode> children = ImmutableList.builder();
            boolean changed = false;
            for (PlanNode child : node.getSources()) {
                PlanNode newChild = child.accept(this, null);
                if (newChild != child) {
                    changed = true;
                }
                children.add(newChild);
            }
            if (!changed) {
                return node;
            }
            return node.replaceChildren(children.build());
        }

        @Override
        public PlanNode visitFilter(FilterNode filter, Void context) {
            if (!(filter.getSource() instanceof TableScanNode)) {
                return filter;
            }

            TableScanNode tableScan = (TableScanNode) filter.getSource();

            Map<String, PrestoColumnHandle> nameToColumnHandlesMapping =
                    tableScan.getAssignments().entrySet().stream()
                            .collect(
                                    Collectors.toMap(
                                            e -> e.getKey().getName(),
                                            e -> (PrestoColumnHandle) e.getValue()));

            RowExpression filterPredicate = filter.getPredicate();

            DomainTranslator.ExtractionResult<Subfield> decomposedFilter =
                    rowExpressionService
                            .getDomainTranslator()
                            .fromPredicate(
                                    session,
                                    filterPredicate,
                                    new SubfieldExtractor(
                                                    functionResolution,
                                                    rowExpressionService.getExpressionOptimizer(),
                                                    session)
                                            .toColumnExtractor());

            // Build paimon predicate presto column.
            TupleDomain<PrestoColumnHandle> entireColumnDomain =
                    decomposedFilter
                            .getTupleDomain()
                            .transform(
                                    subfield ->
                                            subfield.getPath().isEmpty()
                                                    ? subfield.getRootName()
                                                    : null)
                            .transform(nameToColumnHandlesMapping::get);

            // Build paimon predicate column list.
            Map<VariableReferenceExpression, ColumnHandle> assignments = tableScan.getAssignments();
            Optional<List<ColumnHandle>> projectedColumns =
                    extractColumns(filterPredicate, assignments);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



paimon-presto-common/src/main/java/org/apache/paimon/presto/PrestoComputePushdown.java [91:216]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class PrestoComputePushdown implements ConnectorPlanOptimizer {

    private final StandardFunctionResolution functionResolution;
    private final RowExpressionService rowExpressionService;
    private final FunctionMetadataManager functionMetadataManager;
    private final PrestoTransactionManager transactionManager;

    public PrestoComputePushdown(
            StandardFunctionResolution functionResolution,
            RowExpressionService rowExpressionService,
            FunctionMetadataManager functionMetadataManager,
            PrestoTransactionManager transactionManager) {
        this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
        this.rowExpressionService =
                requireNonNull(rowExpressionService, "rowExpressionService is null");
        this.functionMetadataManager =
                requireNonNull(functionMetadataManager, "functionMetadataManager is null");
        this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
    }

    @Override
    public PlanNode optimize(
            PlanNode maxSubplan,
            ConnectorSession session,
            VariableAllocator variableAllocator,
            PlanNodeIdAllocator idAllocator) {

        if (isPaimonPushdownEnabled(session)) {
            return maxSubplan.accept(new Visitor(session, idAllocator), null);
        }
        return maxSubplan;
    }

    private class Visitor extends PlanVisitor<PlanNode, Void> {

        private final ConnectorSession session;
        private final PlanNodeIdAllocator idAllocator;

        public Visitor(ConnectorSession session, PlanNodeIdAllocator idAllocator) {
            this.session = requireNonNull(session, "session is null");
            this.idAllocator = requireNonNull(idAllocator, "idAllocator is null");
        }

        private Optional<List<ColumnHandle>> extractColumns(
                RowExpression expression,
                Map<VariableReferenceExpression, ColumnHandle> assignments) {

            if (expression instanceof VariableReferenceExpression) {
                VariableReferenceExpression variable = (VariableReferenceExpression) expression;
                ColumnHandle columnHandle = assignments.get(variable);
                if (columnHandle != null) {
                    return Optional.of(Collections.singletonList(columnHandle));
                }
                return Optional.empty();
            } else if (expression instanceof CallExpression) {
                CallExpression callExpression = (CallExpression) expression;
                List<ColumnHandle> columns = new ArrayList<>();
                for (RowExpression argument : callExpression.getArguments()) {
                    extractColumns(argument, assignments).ifPresent(columns::addAll);
                }
                return Optional.of(columns);
            }
            return Optional.empty();
        }

        @Override
        public PlanNode visitPlan(PlanNode node, Void context) {
            ImmutableList.Builder<PlanNode> children = ImmutableList.builder();
            boolean changed = false;
            for (PlanNode child : node.getSources()) {
                PlanNode newChild = child.accept(this, null);
                if (newChild != child) {
                    changed = true;
                }
                children.add(newChild);
            }
            if (!changed) {
                return node;
            }
            return node.replaceChildren(children.build());
        }

        @Override
        public PlanNode visitFilter(FilterNode filter, Void context) {
            if (!(filter.getSource() instanceof TableScanNode)) {
                return filter;
            }

            TableScanNode tableScan = (TableScanNode) filter.getSource();

            Map<String, PrestoColumnHandle> nameToColumnHandlesMapping =
                    tableScan.getAssignments().entrySet().stream()
                            .collect(
                                    Collectors.toMap(
                                            e -> e.getKey().getName(),
                                            e -> (PrestoColumnHandle) e.getValue()));

            RowExpression filterPredicate = filter.getPredicate();

            DomainTranslator.ExtractionResult<Subfield> decomposedFilter =
                    rowExpressionService
                            .getDomainTranslator()
                            .fromPredicate(
                                    session,
                                    filterPredicate,
                                    new SubfieldExtractor(
                                                    functionResolution,
                                                    rowExpressionService.getExpressionOptimizer(),
                                                    session)
                                            .toColumnExtractor());

            // Build paimon predicate presto column.
            TupleDomain<PrestoColumnHandle> entireColumnDomain =
                    decomposedFilter
                            .getTupleDomain()
                            .transform(
                                    subfield ->
                                            subfield.getPath().isEmpty()
                                                    ? subfield.getRootName()
                                                    : null)
                            .transform(nameToColumnHandlesMapping::get);

            // Build paimon predicate column list.
            Map<VariableReferenceExpression, ColumnHandle> assignments = tableScan.getAssignments();
            Optional<List<ColumnHandle>> projectedColumns =
                    extractColumns(filterPredicate, assignments);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -