Hive优化器原理与源码解析系列—统计信息之选择性

  /**
   * @param t TableScan操作数
   * @param mq RelMetadataQuery访问Hive收集元数据信息，如记录数、非空记录数、记录长度大小等用于计算成本
   * @param predicate 行表达式，calcite RexNode 相当于SQL where条件部分
   * @return 选择性
   */
public Double getSelectivity(HiveTableScan t, RelMetadataQuery mq, RexNode predicate) {
  if (predicate != null) { //如果谓词不为null，则传递谓词进行估算，否则，即tablescan
    FilterSelectivityEstimator filterSelEstmator = new FilterSelectivityEstimator(t); //使用FilterSelectivityEstimator来选择性
    return filterSelEstmator.estimateSelectivity(predicate);
  }

  return 1.0; //谓词为null，则选择性100%
}

public Double getSelectivity(Join j, RelMetadataQuery mq, RexNode predicate) {
  if (j.getJoinType().equals(JoinRelType.INNER)) {//如果是inner join，返回inner join的选择性，相见下面函数
    return computeInnerJoinSelectivity(j, mq, predicate);
  } else if (j.getJoinType().equals(JoinRelType.LEFT) ||
          j.getJoinType().equals(JoinRelType.RIGHT)) {//如是 左连接 或 右连接 ，分别通过mq获取左右两侧的记录数
    double left = mq.getRowCount(j.getLeft());
    double right = mq.getRowCount(j.getRight());
    double product = left * right;  //左右两边记录数的积
    double innerJoinSelectivity = computeInnerJoinSelectivity(j, mq, predicate);  //再计算此Join如果改成Inner join的选择性
    if (j.getJoinType().equals(JoinRelType.LEFT)) {
      return Math.max(innerJoinSelectivity, left/product);  //如果是leftjion 则inner join选择性和1/right记录数 取最大
    }
    return Math.max(innerJoinSelectivity, right/product);//如果是rightjion 则inner join选择性和1/left记录数 取最大
  }
  return 1.0;
}

ndvCrossProduct = Math.min(mq.getRowCount(j.getLeft()),ndvCrossProduct);

ndvCrossProduct = Math.min(mq.getRowCount(j.getLeft()) * mq.getRowCount(j.getRight()), ndvCrossProduct);

private Double computeInnerJoinSelectivity(Join j, RelMetadataQuery mq, RexNode predicate) {
  Pair<Boolean, RexNode> predInfo =
      getCombinedPredicateForJoin(j, predicate); //返回joinConditon和谓词是否相等的boolean标志及谓词并集
  if (!predInfo.getKey()) { //说明含有谓词，并与joinCondition是不相等的，则FilterSelectivityEstimator估算join的谓词选择性
    return
        new FilterSelectivityEstimator(j).
        estimateSelectivity(predInfo.getValue());
  }

  /**
   * 以下是join，仅有joinCondition而没有谓词即where条件的情况分析：
   */
  RexNode combinedPredicate = predInfo.getValue(); //获取联合的谓词
  JoinPredicateInfo jpi;
  try {
    jpi = JoinPredicateInfo.constructJoinPredicateInfo(j,
        combinedPredicate);    //创建JoinPredicateInfo对象
  } catch (CalciteSemanticException e) {
    throw new RuntimeException(e);
  }
  ImmutableMap.Builder<Integer, Double> colStatMapBuilder = ImmutableMap
      .builder();
  ImmutableMap<Integer, Double> colStatMap;

  int rightOffSet = j.getLeft().getRowType().getFieldCount(); //获取左侧字段个数，索引+1是右侧索引的起点

  // 1. Update Col Stats Map with col stats for columns from left side of
  // Join which are part of join keys
  //使用stats模块功能，更新Join左侧关联字段join keys这些列统计信息映射
  //
  for (Integer ljk : jpi.getProjsFromLeftPartOfJoinKeysInChildSchema()) {//从JoinLeafPredicateInfo对象其Join key在Join Schema中，获取左侧投影列集合·
    colStatMapBuilder.put(ljk,
        HiveRelMdDistinctRowCount.getDistinctRowCount(j.getLeft(), mq, ljk));//取得左侧投影列的，基数，并添加到投影列索引与基数的映射关系
  }

  // 2. Update Col Stats Map with col stats for columns from right side of
  // Join which are part of join keys
  //使用stats模块功能，更新Join右侧关联字段join keys这些列统计信息映射
  for (Integer rjk : jpi.getProjsFromRightPartOfJoinKeysInChildSchema()) {//取得右侧投影列的，基数，并添加到投影列索引与基数的映射关系
    colStatMapBuilder.put(rjk + rightOffSet,  // 这里在左侧投影列索引的基础上
        HiveRelMdDistinctRowCount.getDistinctRowCount(j.getRight(), mq, rjk));
  }
  colStatMap = colStatMapBuilder.build();

  // 3. Walk through the Join Condition Building NDV for selectivity
  // NDV of the join can not exceed the cardinality of cross join.
  //遍历Join Condition关联条件（考虑的是没where条件的模块），来构建非重复记录数的选择性。join的非重复记录数不能超过cross join的基数
  List<JoinLeafPredicateInfo> peLst = jpi.getEquiJoinPredicateElements();
  int noOfPE = peLst.size();
  double ndvCrossProduct = 1;
  if (noOfPE > 0) {
    ndvCrossProduct = exponentialBackoff(peLst, colStatMap); //使用指数后退来计算ndvCrossProduct

    if (j instanceof SemiJoin) { //如果join是SemiJoin类，ndvCrossProduct取值为 ndvCrossProduct 和 join左侧记录数取最小值
      ndvCrossProduct = Math.min(mq.getRowCount(j.getLeft()),
          ndvCrossProduct);
    }else if (j instanceof HiveJoin){ //如果join是Join类，ndvCrossProduct取值为 ndvCrossProduct 和 join左侧记录数*join右侧记录数取最小值
      ndvCrossProduct = Math.min(mq.getRowCount(j.getLeft())
          * mq.getRowCount(j.getRight()), ndvCrossProduct);
    } else {
      throw new RuntimeException("Unexpected Join type: " + j.getClass().getName());
    }
  }

  // 4. Join Selectivity = 1/NDV
  return (1 / ndvCrossProduct);
}

protected double exponentialBackoff(List<JoinLeafPredicateInfo> peLst,
    ImmutableMap<Integer, Double> colStatMap) {
  int noOfPE = peLst.size();
  List<Double> ndvs = new ArrayList<Double>(noOfPE);
  for (int i = 0; i < noOfPE; i++) {
    ndvs.add(getMaxNDVForJoinSelectivity(peLst.get(i), colStatMap));
  }
  Collections.sort(ndvs);
  Collections.reverse(ndvs);//倒排序
  double ndvCrossProduct = 1.0;
  for (int i = 0; i < ndvs.size(); i++) {
    double n = Math.pow(ndvs.get(i), Math.pow(1 / 2.0, i));
    ndvCrossProduct *= n;
  }
  return ndvCrossProduct;
}

  /**
   * 从投影列集合中选列最大基数
   *
   * @param colStatMap 投影列序数，基数（非重复列记录数）映射关系
   * @param projectionSet 投影列序数集合
   * @param defaultMaxNDV 默认最大基数
   * @return 返回投影列中最大的基数
   */
  private static Double getMaxNDVFromProjections(Map<Integer, Double> colStatMap,
      Set<Integer> projectionSet, Double defaultMaxNDV) {
    Double colNDV = null;
    Double maxNDVSoFar = defaultMaxNDV;

    for (Integer projIndx : projectionSet) { //遍历投影列集合，比较投影列序数，基数（非重复列记录数）映射关系，最大基数并返回
      colNDV = colStatMap.get(projIndx);
      if (colNDV > maxNDVSoFar)
        maxNDVSoFar = colNDV;//取最大NDV
    }

    return maxNDVSoFar;
  }

}

private static Double getMaxNDVForJoinSelectivity(JoinLeafPredicateInfo jlpi,
    ImmutableMap<Integer, Double> colStatMap) {
  Double maxNDVSoFar = 1.0; //初始最大NDV为1

  maxNDVSoFar = getMaxNDVFromProjections(colStatMap,
      jlpi.getProjsFromLeftPartOfJoinKeysInJoinSchema(), maxNDVSoFar);//取得左侧投影列集合中最大NDV
  maxNDVSoFar = getMaxNDVFromProjections(colStatMap,
      jlpi.getProjsFromRightPartOfJoinKeysInJoinSchema(), maxNDVSoFar);//使用左侧投影列集合中最大NDV作为参考，再选取左右两侧中最大的NDV

  return maxNDVSoFar;
}

cross product是通过所有连接谓词中最大NDV乘以其余连接谓词降级NDV。
NDV是使用log函数来降级的。ndvCrossProduct圈定了Join的范围，cross product确保NDV不会超过最坏情况下的join

protected double logSmoothing(List<JoinLeafPredicateInfo> peLst, ImmutableMap<Integer, Double> colStatMap) {
  int noOfPE = peLst.size();
  double ndvCrossProduct = getMaxNDVForJoinSelectivity(peLst.get(0), colStatMap);
  if (noOfPE > 1) {
    double maxNDVSoFar = ndvCrossProduct;
    double ndvToBeSmoothed;
    double tmpNDV;

    for (int i = 1; i < noOfPE; i++) {
      tmpNDV = getMaxNDVForJoinSelectivity(peLst.get(i), colStatMap);
      if (tmpNDV > maxNDVSoFar) {
        ndvToBeSmoothed = maxNDVSoFar;
        maxNDVSoFar = tmpNDV;
        ndvCrossProduct = (ndvCrossProduct / ndvToBeSmoothed) * tmpNDV;
      } else {
        ndvToBeSmoothed = tmpNDV;
      }
      // TODO: revisit the fence
      if (ndvToBeSmoothed > 3)
        ndvCrossProduct *= Math.log(ndvToBeSmoothed);//使用log函数进行平滑计算
      else
        ndvCrossProduct *= ndvToBeSmoothed;
    }
  }
  return ndvCrossProduct;
}