public static Optional<LocalDynamicFilter> create(AbstractJoinNode planNode, int partitionCount)
    {
        Set<String> joinDynamicFilters = planNode.getDynamicFilters().keySet();
        List<FilterNode> filterNodes = PlanNodeSearcher
                .searchFrom(planNode.getProbe())
                .where(PlannerUtils::isFilterAboveTableScan)
                .findAll();

        // Mapping from probe-side dynamic filters' IDs to their matching probe variables.
        ImmutableMultimap.Builder<String, DynamicFilterPlaceholder> probeVariablesBuilder = ImmutableMultimap.builder();
        for (FilterNode filterNode : filterNodes) {
            DynamicFilterExtractResult extractResult = extractDynamicFilters(filterNode.getPredicate());
            for (DynamicFilterPlaceholder placeholder : extractResult.getDynamicConjuncts()) {
                if (placeholder.getInput() instanceof VariableReferenceExpression) {
                    // Add descriptors that match the local dynamic filter (from the current join node).
                    if (joinDynamicFilters.contains(placeholder.getId())) {
                        probeVariablesBuilder.put(placeholder.getId(), placeholder);
                    }
                }
            }
        }

        Multimap<String, DynamicFilterPlaceholder> probeVariables = probeVariablesBuilder.build();
        PlanNode buildNode = planNode.getBuild();
        Map<String, Integer> buildChannels = planNode.getDynamicFilters().entrySet().stream()
                // Skip build channels that don't match local probe dynamic filters.
                .filter(entry -> probeVariables.containsKey(entry.getKey()))
                .collect(toMap(
                        // Dynamic filter ID
                        Map.Entry::getKey,
                        // Build-side channel index
                        entry -> {
                            VariableReferenceExpression buildVariable = entry.getValue();
                            int buildChannelIndex = buildNode.getOutputVariables().indexOf(buildVariable);
                            verify(buildChannelIndex >= 0);
                            return buildChannelIndex;
                        }));

        if (buildChannels.isEmpty()) {
            return Optional.empty();
        }
        return Optional.of(new LocalDynamicFilter(probeVariables, buildChannels, partitionCount));
    }
@Test
    public void testCreateDistributedJoin()
    {
        Session session = Session.builder(getQueryRunner().getDefaultSession())
                .setSystemProperty(JOIN_DISTRIBUTION_TYPE, "PARTITIONED")
                .build();
        SubPlan subplan = subplan(
                "SELECT count() FROM nation, region WHERE nation.regionkey = region.regionkey " + "AND region.comment = 'abc'",
                Optimizer.PlanStage.OPTIMIZED_AND_VALIDATED,
                false,
                session);
        JoinNode joinNode = searchJoins(subplan.getChildren().get(0).getFragment()).findOnlyElement();
        assertFalse(joinNode.getDynamicFilters().isEmpty());
        assertEquals(LocalDynamicFilter.create(joinNode, 1), Optional.empty());
    }