@Override
    public Result apply(ApplyNode applyNode, Captures captures, Context context)
    {
        if (applyNode.getMayParticipateInAntiJoin()) {
            return Result.empty();
        }

        Assignments subqueryAssignments = applyNode.getSubqueryAssignments();
        if (subqueryAssignments.size() != 1) {
            return Result.empty();
        }

        RowExpression expression = getOnlyElement(subqueryAssignments.getExpressions());
        if (!(expression instanceof InSubqueryExpression)) {
            return Result.empty();
        }
        InSubqueryExpression inPredicate = (InSubqueryExpression) expression;

        VariableReferenceExpression inPredicateOutputVariable = getOnlyElement(subqueryAssignments.getVariables());

        PlanNode leftInput = applyNode.getInput();
        // Add unique id column if the set of columns do not form a unique key already
        if (!((GroupReference) leftInput).getLogicalProperties().isPresent() ||
                !((GroupReference) leftInput).getLogicalProperties().get().isDistinct(ImmutableSet.copyOf(leftInput.getOutputVariables()))) {
            VariableReferenceExpression uniqueKeyVariable = context.getVariableAllocator().newVariable("unique", BIGINT);
            leftInput = new AssignUniqueId(
                    applyNode.getSourceLocation(),
                    context.getIdAllocator().getNextId(),
                    leftInput,
                    uniqueKeyVariable);
        }

        VariableReferenceExpression leftVariableReference = inPredicate.getValue();
        VariableReferenceExpression rightVariableReference = inPredicate.getSubquery();

        JoinNode innerJoin = new JoinNode(
                applyNode.getSourceLocation(),
                context.getIdAllocator().getNextId(),
                JoinType.INNER,
                leftInput,
                applyNode.getSubquery(),
                ImmutableList.of(new EquiJoinClause(
                        leftVariableReference,
                        rightVariableReference)),
                ImmutableList.<VariableReferenceExpression>builder()
                        .addAll(leftInput.getOutputVariables())
                        .build(),
                Optional.empty(),
                Optional.empty(),
                Optional.empty(),
                Optional.empty(),
                ImmutableMap.of());

        AggregationNode distinctNode = new AggregationNode(
                innerJoin.getSourceLocation(),
                context.getIdAllocator().getNextId(),
                innerJoin,
                ImmutableMap.of(),
                singleGroupingSet(ImmutableList.<VariableReferenceExpression>builder()
                        .addAll(innerJoin.getOutputVariables())
                        .build()),
                ImmutableList.of(),
                SINGLE,
                Optional.empty(),
                Optional.empty(),
                Optional.empty());

        ImmutableList<VariableReferenceExpression> referencedOutputs = ImmutableList.<VariableReferenceExpression>builder()
                .addAll(applyNode.getInput().getOutputVariables())
                .add(inPredicateOutputVariable)
                .build();

        ProjectNode finalProjectNdde = new ProjectNode(
                context.getIdAllocator().getNextId(),
                distinctNode,
                Assignments.builder()
                        .putAll(identityAssignments(distinctNode.getOutputVariables()))
                        .put(inPredicateOutputVariable, TRUE_CONSTANT)
                        .build()
                        .filter(referencedOutputs));

        return Result.ofPlanNode(finalProjectNdde);
    }
@Test
    public void testDoesNotFiresForInPredicateThatMayParticipateInAntiJoin()
    {
        tester().assertThat(new TransformUncorrelatedInPredicateSubqueryToDistinctInnerJoin())
                .on(p -> p.apply(
                        assignment(
                                p.variable("x"),
                                inSubquery(p.variable("y"), p.variable("z"))),
                        emptyList(),
                        p.values(p.variable("y")),
                        p.values(p.variable("z")),
                        true))
                .doesNotFire();
    }