Hive优化器原理与源码解析系列--优化规则HiveAggregateProjectMergeRule(十六)

protected Aggregate(RelOptCluster cluster,
                    RelTraitSet traitSet,
                    RelNode input,
                    ImmutableBitSet groupSet,
                    java.util.List<ImmutableBitSet> groupSets,
                    java.util.List<AggregateCall> aggCalls)
参数说明:
cluster - 当前集群，提供执行环境
traitSet - 特征集合
input - 输入RelNode
groupSet - GroupSet变量
groupSets - GroupSets元素列表
aggCalls - 调用汇总函数的集合

@Override
public boolean matches(RelOptRuleCall call) {
  final Aggregate aggregate = call.rel(0); //获取root根节点表达式
  for (AggregateCall aggCall : aggregate.getAggCallList()) {
    if (aggCall.getAggregation().equals(HiveGroupingID.INSTANCE)) {//判断是否含有GroupID
      return false;
    }
  }
  return super.matches(call);
}

SELECT 
month,
day,
COUNT(DISTINCT cookieid) AS uv,
GROUPING__ID
FROM tab_test
GROUP BY month,day
GROUPING SETS (month,day,(month,day))
ORDER BY GROUPING__ID;

结果：
month         day             uv      GROUPING__ID
------------------------------------------------
2015-03       NULL            5       1
2015-04       NULL            6       1
NULL          2015-03-10      4       2
NULL          2015-03-12      1       2
NULL          2015-04-12      2       2
2015-03       2015-03-10      4       3
2015-03       2015-03-12      1       3
2015-04       2015-04-12      2       3

等价于
SELECT month,NULL,COUNT(DISTINCT cookieid) AS uv,1 AS GROUPING__ID FROM tab_test GROUP BY month
UNION ALL
SELECT NULL,day,COUNT(DISTINCT cookieid) AS uv,2 AS GROUPING__ID FROM tab_test GROUP BY day
UNION ALL
SELECT month,day,COUNT(DISTINCT cookieid) AS uv,3 AS GROUPING__ID FROM tab_test GROUP BY month,day
说明：grouping sets 只会根据，sets集合内每个元素单独分组：month、day、（month，day）三个分组
注意：group by中字段集合 要 包含 grouping sets（）集合字段，否则会报错，即{group by} >={grouping sets}

@Override
public void onMatch(RelOptRuleCall call) {
  final HiveAggregate aggregate = call.rel(0);// 根表达式root expression 为Aggregate
  final HiveProject project = call.rel(1); //下一个表达式为Project
  RelNode x = apply(aggregate, project);    //两个操作应用到一个RelNode
  if (x != null) {
    call.transformTo(x);//调用转换
  }
}

public static RelNode apply(HiveAggregate aggregate,
    HiveProject project)

RexInputRef：（序号,字段数据类型）代表 一个字段

* Input #0: EMP(EMPNO, ENAME, DEPTNO) and
* Input #1: DEPT(DEPTNO AS DEPTNO2, DNAME)
  
字段分别如下：

* Field #0: EMPNO
* Field #1: ENAME
* Field #2: DEPTNO (from EMP)
* Field #3: DEPTNO2 (from DEPT)
* Field #4: DNAME

for (int key : aggregate.getGroupSet()) {//Returns a bit set of the grouping fields  （ 如上述：grouping sets(cur_stt,crt_tim) ）
  final RexNode rex = project.getProjects().get(key);  //project.getProjects()返回类型：List<RexNode>  //select 1,2,sum(a) from t group by 1,2
  if (rex instanceof RexInputRef) { //判断Project投影的行表达式，是一个字段的引用，而不是函数之类的
    final int newKey = ((RexInputRef) rex).getIndex(); //取出字段引用的Ref的字段序号。
    newKeys.add(newKey);
    map.put(key, newKey);   //初始化groupset字段索引与投影中字段索引的映射关系
  } else {
    // Cannot handle "GROUP BY expression"
    return null;
  }
}

for (AggregateCall aggregateCall : aggregate.getAggCallList()) {//遍历调用汇总函数，函数列表
  final ImmutableList.Builder<Integer> newArgs = ImmutableList.builder();
  for (int arg : aggregateCall.getArgList()) {//遍历 每个汇总函数内的参数，并到投影中确认，判断是否引用到字段，并添加到newArgs列表中，否则返回为null
    final RexNode rex = project.getProjects().get(arg); // 如果在Project投影中，没有找到则返回null或返回的不是字段引用，最终结果返回null，则会跳出优化
    if (rex instanceof RexInputRef) {
      newArgs.add(((RexInputRef) rex).getIndex());
    } else {
      // Cannot handle "AGG(expression)"
      return null;
    }
  }

  RelNode rel = newAggregate;
  if (!newKeys.equals(newGroupSet.asList())) { //判断这两个列表是否相等，如果不相等，则进行遍历newKeys索引，并查找对应newGroupSet索引位置，添加到postList中。
    final List<Integer> posList = Lists.newArrayList();
    for (int newKey : newKeys) {
      posList.add(newGroupSet.indexOf(newKey));
    }
    if (aggregate.indicator) {//如果indicator为true
      for (int newKey : newKeys) {
        posList.add(aggregate.getGroupCount() + newGroupSet.indexOf(newKey));//在分组字段个数的基础上+索引
      }
    }
    for (int i = newAggregate.getGroupCount()
                 + newAggregate.getIndicatorCount();
         i < newAggregate.getRowType().getFieldCount(); i++) {
      posList.add(i);
    }
    rel = HiveRelOptUtil.createProject(
        HiveRelFactories.HIVE_BUILDER.create(aggregate.getCluster(), null),
        rel, posList);// 这里合并最要的一步：使用new Aggregate和posList列表创建一个new Project投影。这里完成了Aggregate和Project合并的操作作为一个RelNode。
  }

  return rel;
}