@Override
    public void onMatch(RelOptRuleCall call) {
      final LogicalCorrelate correlate = call.rel(0);
      final RelNode left = call.rel(1);
      final Project aggOutputProject = call.rel(2);
      final Aggregate aggregate = call.rel(3);
      final Project aggInputProject = call.rel(4);
      RelNode right = call.rel(5);
      final RelOptCluster cluster = correlate.getCluster();

      setCurrent(call.getPlanner().getRoot(), correlate);

      // check for this pattern
      // The pattern matching could be simplified if rules can be applied
      // during decorrelation,
      //
      // CorrelateRel(left correlation, condition = true)
      //   LeftInputRel
      //   Project-A (a RexNode)
      //     Aggregate (groupby (0), agg0(), agg1()...)
      //       Project-B (references coVar)
      //         rightInputRel

      // check aggOutputProject projects only one expression
      final List<RexNode> aggOutputProjects = aggOutputProject.getProjects();
      if (aggOutputProjects.size() != 1) {
        return;
      }

      if(correlate.getJoinType() != JoinRelType.LEFT) {
        return;
      }

      final JoinRelType joinType = correlate.getJoinType();
      // corRel.getCondition was here, however Correlate was updated so it
      // never includes a join condition. The code was not modified for brevity.
      RexNode joinCond = rexBuilder.makeLiteral(true);
      if ((joinType != JoinRelType.LEFT)
              || (joinCond != rexBuilder.makeLiteral(true))) {
        return;
      }

      // check that the agg is on the entire input
      if (!aggregate.getGroupSet().isEmpty()) {
        return;
      }

      final List<RexNode> aggInputProjects = aggInputProject.getProjects();

      final List<AggregateCall> aggCalls = aggregate.getAggCallList();
      final Set<Integer> isCountStar = Sets.newHashSet();

      // mark if agg produces count(*) which needs to reference the
      // nullIndicator after the transformation.
      int k = -1;
      for (AggregateCall aggCall : aggCalls) {
        ++k;
        if ((aggCall.getAggregation() instanceof SqlCountAggFunction)
                && (aggCall.getArgList().size() == 0)) {
          isCountStar.add(k);
        }
      }

      if ((right instanceof Filter)
              && cm.mapRefRelToCorRef.containsKey(right)) {
        // rightInputRel has this shape:
        //
        //       Filter (references corvar)
        //         FilterInputRel
        Filter filter = (Filter) right;
        right = filter.getInput();

        assert right instanceof HepRelVertex;
        right = ((HepRelVertex) right).getCurrentRel();

        // check filter input contains no correlation
        if (RelOptUtil.getVariablesUsed(right).size() > 0) {
          return;
        }

        // check filter condition type First extract the correlation out
        // of the filter

        // First breaking up the filter conditions into equality
        // comparisons between rightJoinKeys(from the original
        // filterInputRel) and correlatedJoinKeys. correlatedJoinKeys
        // can only be RexFieldAccess, while rightJoinKeys can be
        // expressions. These comparisons are AND'ed together.
        List<RexNode> rightJoinKeys = Lists.newArrayList();
        List<RexNode> tmpCorrelatedJoinKeys = Lists.newArrayList();
        HiveRelOptUtil.splitCorrelatedFilterCondition(
                filter,
                rightJoinKeys,
                tmpCorrelatedJoinKeys,
                true);

        // make sure the correlated reference forms a unique key check
        // that the columns referenced in these comparisons form an
        // unique key of the leftInputRel
        List<RexFieldAccess> correlatedJoinKeys = Lists.newArrayList();
        List<RexInputRef> correlatedInputRefJoinKeys = Lists.newArrayList();
        for (RexNode joinKey : tmpCorrelatedJoinKeys) {
          assert joinKey instanceof RexFieldAccess;
          correlatedJoinKeys.add((RexFieldAccess) joinKey);
          RexNode correlatedInputRef =
                  removeCorrelationExpr(joinKey, false);
          assert correlatedInputRef instanceof RexInputRef;
          correlatedInputRefJoinKeys.add(
                  (RexInputRef) correlatedInputRef);
        }

        // check that the columns referenced in rightJoinKeys form an
        // unique key of the filterInputRel
        if (correlatedInputRefJoinKeys.isEmpty()) {
          return;
        }

        // The join filters out the nulls.  So, it's ok if there are
        // nulls in the join keys.
        final RelMetadataQuery mq = call.getMetadataQuery();
        if (!RelMdUtil.areColumnsDefinitelyUniqueWhenNullsFiltered(mq, left,
                correlatedInputRefJoinKeys)) {
          //SQL2REL_LOGGER.fine(correlatedJoinKeys.toString()
           //       + "are not unique keys for "
            //      + left.toString());
          return;
        }

        // check cor var references are valid
        if (!checkCorVars(correlate,
                aggInputProject,
                filter,
                correlatedJoinKeys)) {
          return;
        }

        // Rewrite the above plan:
        //
        // CorrelateRel(left correlation, condition = true)
        //   LeftInputRel
        //   Project-A (a RexNode)
        //     Aggregate (groupby(0), agg0(),agg1()...)
        //       Project-B (may reference coVar)
        //         Filter (references corVar)
        //           RightInputRel (no correlated reference)
        //

        // to this plan:
        //
        // Project-A' (all gby keys + rewritten nullable ProjExpr)
        //   Aggregate (groupby(all left input refs)
        //                 agg0(rewritten expression),
        //                 agg1()...)
        //     Project-B' (rewriten original projected exprs)
        //       Join(replace corvar w/ input ref from LeftInputRel)
        //         LeftInputRel
        //         RightInputRel
        //

        // In the case where agg is count(*) or count($corVar), it is
        // changed to count(nullIndicator).
        // Note:  any non-nullable field from the RHS can be used as
        // the indicator however a "true" field is added to the
        // projection list from the RHS for simplicity to avoid
        // searching for non-null fields.
        //
        // Project-A' (all gby keys + rewritten nullable ProjExpr)
        //   Aggregate (groupby(all left input refs),
        //                 count(nullIndicator), other aggs...)
        //     Project-B' (all left input refs plus
        //                    the rewritten original projected exprs)
        //       Join(replace corvar to input ref from LeftInputRel)
        //         LeftInputRel
        //         Project (everything from RightInputRel plus
        //                     the nullIndicator "true")
        //           RightInputRel
        //

        // first change the filter condition into a join condition
        joinCond =
                removeCorrelationExpr(filter.getCondition(), false);
      } else if (cm.mapRefRelToCorRef.containsKey(aggInputProject)) {
        // check rightInputRel contains no correlation
        if (RelOptUtil.getVariablesUsed(right).size() > 0) {
          return;
        }

        // check cor var references are valid
        if (!checkCorVars(correlate, aggInputProject, null, null)) {
          return;
        }

        int nFields = left.getRowType().getFieldCount();
        ImmutableBitSet allCols = ImmutableBitSet.range(nFields);

        // leftInputRel contains unique keys
        // i.e. each row is distinct and can group by on all the left
        // fields
        final RelMetadataQuery mq = call.getMetadataQuery();
        if (!RelMdUtil.areColumnsDefinitelyUnique(mq, left, allCols)) {
          //SQL2REL_LOGGER.fine("There are no unique keys for " + left);
          return;
        }
        //
        // Rewrite the above plan:
        //
        // CorrelateRel(left correlation, condition = true)
        //   LeftInputRel
        //   Project-A (a RexNode)
        //     Aggregate (groupby(0), agg0(), agg1()...)
        //       Project-B (references coVar)
        //         RightInputRel (no correlated reference)
        //

        // to this plan:
        //
        // Project-A' (all gby keys + rewritten nullable ProjExpr)
        //   Aggregate (groupby(all left input refs)
        //                 agg0(rewritten expression),
        //                 agg1()...)
        //     Project-B' (rewriten original projected exprs)
        //       Join (LOJ cond = true)
        //         LeftInputRel
        //         RightInputRel
        //

        // In the case where agg is count($corVar), it is changed to
        // count(nullIndicator).
        // Note:  any non-nullable field from the RHS can be used as
        // the indicator however a "true" field is added to the
        // projection list from the RHS for simplicity to avoid
        // searching for non-null fields.
        //
        // Project-A' (all gby keys + rewritten nullable ProjExpr)
        //   Aggregate (groupby(all left input refs),
        //                 count(nullIndicator), other aggs...)
        //     Project-B' (all left input refs plus
        //                    the rewritten original projected exprs)
        //       Join(replace corvar to input ref from LeftInputRel)
        //         LeftInputRel
        //         Project (everything from RightInputRel plus
        //                     the nullIndicator "true")
        //           RightInputRel
      } else {
        return;
      }

      RelDataType leftInputFieldType = left.getRowType();
      int leftInputFieldCount = leftInputFieldType.getFieldCount();
      int joinOutputProjExprCount =
              leftInputFieldCount + aggInputProjects.size() + 1;

      right =
              createProjectWithAdditionalExprs(right,
                      ImmutableList.of(
                              Pair.<RexNode, String>of(rexBuilder.makeLiteral(true),
                                      "nullIndicator")));

      Join join = (Join) relBuilder.push(left).push(right)
          .join(joinType, joinCond).build();

      // To the consumer of joinOutputProjRel, nullIndicator is located
      // at the end
      int nullIndicatorPos = join.getRowType().getFieldCount() - 1;

      RexInputRef nullIndicator =
              new RexInputRef(
                      nullIndicatorPos,
                      cluster.getTypeFactory().createTypeWithNullability(
                              join.getRowType().getFieldList()
                                      .get(nullIndicatorPos).getType(),
                              true));

      // first project all group-by keys plus the transformed agg input
      List<RexNode> joinOutputProjects = Lists.newArrayList();

      // LOJ Join preserves LHS types
      for (int i = 0; i < leftInputFieldCount; i++) {
        joinOutputProjects.add(
                rexBuilder.makeInputRef(
                        leftInputFieldType.getFieldList().get(i).getType(), i));
      }

      for (RexNode aggInputProjExpr : aggInputProjects) {
        joinOutputProjects.add(
                removeCorrelationExpr(aggInputProjExpr,
                        joinType.generatesNullsOnRight(),
                        nullIndicator));
      }

      joinOutputProjects.add(
              rexBuilder.makeInputRef(join, nullIndicatorPos));

      RelNode joinOutputProject = RelOptUtil.createProject(
          join, joinOutputProjects, null, false, relBuilder);

      // nullIndicator is now at a different location in the output of
      // the join
      nullIndicatorPos = joinOutputProjExprCount - 1;

      final int groupCount = leftInputFieldCount;

      List<AggregateCall> newAggCalls = Lists.newArrayList();
      k = -1;
      for (AggregateCall aggCall : aggCalls) {
        ++k;
        final List<Integer> argList;

        if (isCountStar.contains(k)) {
          // this is a count(*), transform it to count(nullIndicator)
          // the null indicator is located at the end
          argList = Collections.singletonList(nullIndicatorPos);
        } else {
          argList = Lists.newArrayList();

          for (int aggArg : aggCall.getArgList()) {
            argList.add(aggArg + groupCount);
          }
        }

        int filterArg = aggCall.filterArg < 0 ? aggCall.filterArg
                : aggCall.filterArg + groupCount;
        newAggCalls.add(
                aggCall.adaptTo(joinOutputProject, argList, filterArg,
                        aggregate.getGroupCount(), groupCount));
      }

      ImmutableBitSet groupSet =
              ImmutableBitSet.range(groupCount);
      Aggregate newAggregate =
          (Aggregate) relBuilder.push(joinOutputProject)
              .aggregate(relBuilder.groupKey(groupSet, null), newAggCalls).build();

      List<RexNode> newAggOutputProjectList = Lists.newArrayList();
      for (int i : groupSet) {
        newAggOutputProjectList.add(
                rexBuilder.makeInputRef(newAggregate, i));
      }

      RexNode newAggOutputProjects =
              removeCorrelationExpr(aggOutputProjects.get(0), false);
      newAggOutputProjectList.add(
              rexBuilder.makeCast(
                      cluster.getTypeFactory().createTypeWithNullability(
                              newAggOutputProjects.getType(),
                              true),
                      newAggOutputProjects));

      RelNode newAggOutputProject = RelOptUtil.createProject(
          newAggregate, newAggOutputProjectList, null, false, relBuilder);

      call.transformTo(newAggOutputProject);

      removeCorVarFromTree(correlate);
    }
Planner rule that removes correlations for scalar aggregates.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveRelDecorrelator.java