presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java/LocalExecutionPlanner/createDynamicFilter
class LocalExecutionPlanner:        private Optional<LocalDynamicFilter> createDynamicFilter(JoinNode node, LocalExecutionPlanContext context, int partitionCount)
        {
            if (!isEnableDynamicFiltering(context.getSession())) {
                return Optional.empty();
            }
            if (node.getDynamicFilters().isEmpty()) {
                return Optional.empty();
            }
            log.debug("[Join] Dynamic filters: %s", node.getDynamicFilters());
            LocalDynamicFiltersCollector collector = context.getDynamicFiltersCollector();
            return LocalDynamicFilter
                    .create(metadata, node, partitionCount)
                    .map(filter -> {
                        // Intersect dynamic filters' predicates when they become ready,
                        // in order to support multiple join nodes in the same plan fragment.
                        addSuccessCallback(filter.getResultFuture(), collector::intersect);
                        return filter;
                    });
        }