@Override
  public void onMatch(RelOptRuleCall call) {
    final Aggregate agg = call.rel(aggregateIndex);
    final Union union = call.rel(1);
    final RelBuilder relBuilder = call.builder();
    final RexBuilder rexBuilder = relBuilder.getRexBuilder();

    // 1) First branch is query, second branch is MV
    RelNode joinLeftInput = union.getInput(1);
    final T joinRightInput = createJoinRightInput(call);
    if (joinRightInput == null) {
      return;
    }

    // 2) Introduce a Project on top of MV scan having all columns from the view plus a boolean literal which indicates
    // whether the row with the key values coming from the joinRightInput exists in the view:
    // - true means exist
    // - null means not exists
    // Project also needed to encapsulate the view scan by a subquery -> this is required by
    // CalcitePlanner.fixUpASTAggregateInsertIncrementalRebuild
    // CalcitePlanner.fixUpASTAggregateInsertDeleteIncrementalRebuild
    List<RexNode> mvCols = new ArrayList<>(joinLeftInput.getRowType().getFieldCount());
    for (int i = 0; i < joinLeftInput.getRowType().getFieldCount(); ++i) {
      mvCols.add(rexBuilder.makeInputRef(
              joinLeftInput.getRowType().getFieldList().get(i).getType(), i));
    }
    mvCols.add(rexBuilder.makeLiteral(true));

    joinLeftInput = relBuilder
            .push(joinLeftInput)
            .project(mvCols)
            .build();

    // 3) Build conditions for join and start adding
    // expressions for project operator
    List<RexNode> projExprs = new ArrayList<>();
    List<RexNode> joinConjs = new ArrayList<>();
    int groupCount = agg.getGroupCount();
    int totalCount = agg.getGroupCount() + agg.getAggCallList().size();
    for (int leftPos = 0, rightPos = totalCount + 1;
         leftPos < groupCount; leftPos++, rightPos++) {
      RexNode leftRef = rexBuilder.makeInputRef(
          joinLeftInput.getRowType().getFieldList().get(leftPos).getType(), leftPos);
      RexNode rightRef = rexBuilder.makeInputRef(
          joinRightInput.rightInput.getRowType().getFieldList().get(leftPos).getType(), rightPos);
      projExprs.add(rightRef);

      RexNode nsEqExpr = rexBuilder.makeCall(SqlStdOperatorTable.IS_NOT_DISTINCT_FROM,
              ImmutableList.of(leftRef, rightRef));
      joinConjs.add(nsEqExpr);
    }

    // 4) Create join node
    RexNode joinCond = RexUtil.composeConjunction(rexBuilder, joinConjs);
    RelNode join = relBuilder
            .push(joinLeftInput)
            .push(joinRightInput.rightInput)
            .join(JoinRelType.RIGHT, joinCond)
            .build();

    // 5) Add the expressions that correspond to the aggregation
    // functions
    for (int i = 0, leftPos = groupCount, rightPos = totalCount + 1 + groupCount;
         leftPos < totalCount; i++, leftPos++, rightPos++) {
      // case when source.s is null and mv2.s is null then null
      // case when source.s IS null then mv2.s
      // case when mv2.s IS null then source.s
      // else source.s + mv2.s end
      RexNode leftRef = rexBuilder.makeInputRef(
          joinLeftInput.getRowType().getFieldList().get(leftPos).getType(), leftPos);
      RexNode rightRef = rexBuilder.makeInputRef(
          joinRightInput.rightInput.getRowType().getFieldList().get(leftPos).getType(), rightPos);
      // Generate SQLOperator for merging the aggregations
      SqlAggFunction aggCall = agg.getAggCallList().get(i).getAggregation();
      RexNode elseReturn = createAggregateNode(aggCall, leftRef, rightRef, rexBuilder);
      // According to SQL standard (and Hive) Aggregate functions eliminates null values however operators used in
      // elseReturn expressions returns null if one of their operands is null
      // hence we need a null check of both operands.
      // Note: If both are null, we will fall into branch    WHEN leftNull THEN rightRef
      RexNode leftNull = rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL, leftRef);
      RexNode rightNull = rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL, rightRef);
      RexNode caseExpression = rexBuilder.makeCall(SqlStdOperatorTable.CASE,
              leftNull, rightRef,
              rightNull, leftRef,
              elseReturn);
      RexNode cast = rexBuilder.makeCast(
              call.rel(0).getRowType().getFieldList().get(projExprs.size()).getType(), caseExpression);
      projExprs.add(cast);
    }

    int flagIndex = joinLeftInput.getRowType().getFieldCount() - 1;
    RexNode flagNode = rexBuilder.makeInputRef(
            join.getRowType().getFieldList().get(flagIndex).getType(), flagIndex);

    // 6) Build plan
    RelNode newNode = relBuilder
        .push(join)
        .filter(createFilterCondition(joinRightInput, flagNode, projExprs, relBuilder))
        .project(projExprs)
        .build();
    call.transformTo(newNode);
  }
This class implements the main algorithm of rewriting an MV maintenance plan to incremental.
Subclasses are represents Calcite rules which performs a rewriting to prepare the plan for incremental
view maintenance in case there exist aggregation operator, so we can
avoid the INSERT OVERWRITE and use a MERGE statement instead.

See subclasses for examples:
  {@link HiveAggregateInsertIncrementalRewritingRule},
  {@link HiveAggregateInsertDeleteIncrementalRewritingRule}

@param <T> Should be a class that wraps the right input of the top right outer join: the union branch
          which represents the plan which produces the delta records since the last view rebuild.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/views/HiveAggregateIncrementalRewritingRuleBase.java