presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java/LocalDynamicFilter/create
class LocalDynamicFilter:    public static Optional<LocalDynamicFilter> create(Metadata metadata, JoinNode planNode, int partitionCount)
    {
        Set<String> joinDynamicFilters = planNode.getDynamicFilters().keySet();
        List<FilterNode> filterNodes = PlanNodeSearcher
                .searchFrom(planNode.getLeft())
                .where(LocalDynamicFilter::isFilterAboveTableScan)
                .findAll();

        // Mapping from probe-side dynamic filters' IDs to their matching probe symbols.
        ImmutableMultimap.Builder<String, Symbol> probeSymbolsBuilder = ImmutableMultimap.builder();
        for (FilterNode filterNode : filterNodes) {
            DynamicFilters.ExtractResult extractResult = extractDynamicFilters(metadata, filterNode.getPredicate());
            for (Descriptor descriptor : extractResult.getDynamicConjuncts()) {
                if (descriptor.getInput() instanceof SymbolReference) {
                    // Add descriptors that match the local dynamic filter (from the current join node).
                    if (joinDynamicFilters.contains(descriptor.getId())) {
                        Symbol probeSymbol = Symbol.from(descriptor.getInput());
                        log.debug("Adding dynamic filter %s: %s", descriptor, probeSymbol);
                        probeSymbolsBuilder.put(descriptor.getId(), probeSymbol);
                    }
                }
            }
        }

        Multimap<String, Symbol> probeSymbols = probeSymbolsBuilder.build();
        PlanNode buildNode = planNode.getRight();
        Map<String, Integer> buildChannels = planNode.getDynamicFilters().entrySet().stream()
                // Skip build channels that don't match local probe dynamic filters.
                .filter(entry -> probeSymbols.containsKey(entry.getKey()))
                .collect(toMap(
                        // Dynamic filter ID
                        entry -> entry.getKey(),
                        // Build-side channel index
                        entry -> {
                            Symbol buildSymbol = entry.getValue();
                            int buildChannelIndex = buildNode.getOutputSymbols().indexOf(buildSymbol);
                            verify(buildChannelIndex >= 0);
                            return buildChannelIndex;
                        }));

        if (buildChannels.isEmpty()) {
            return Optional.empty();
        }
        return Optional.of(new LocalDynamicFilter(probeSymbols, buildChannels, partitionCount));
    }