Hive优化器原理与源码解析系列—CBO成本模型CostModel(一)

public static final HiveCost INFINITY = new HiveCost(
                                  Double.POSITIVE_INFINITY,
                                  Double.POSITIVE_INFINITY,
                                  Double.POSITIVE_INFINITY)

public static final HiveCost HUGE = new HiveCost(
                             Double.MAX_VALUE,
                             Double.MAX_VALUE,
                             Double.MAX_VALUE)

public static final HiveCost ZERO = new HiveCost(0.0, 0.0, 0.0)

 public static final HiveCost ZERO = new HiveCost(1.0, 1.0, 1.0)

public double divideBy(RelOptCost cost) { 
  // Compute the geometric average of the ratios of all of the factors
  // which are non-zero and finite.
  double d = 1;
  double n = 0;
  if ((this.rowCount != 0) && !Double.isInfinite(this.rowCount) && (cost.getRows() != 0)
      && !Double.isInfinite(cost.getRows())) {
    d *= this.rowCount / cost.getRows();//判断为非无穷大，并记录数不为0的情况下进行累乘计算
    ++n;
  }
  if ((this.cpu != 0) && !Double.isInfinite(this.cpu) && (cost.getCpu() != 0)
      && !Double.isInfinite(cost.getCpu())) {
    d *= this.cpu / cost.getCpu();
    ++n;
  }
  if ((this.io != 0) && !Double.isInfinite(this.io) && (cost.getIo() != 0)
      && !Double.isInfinite(cost.getIo())) {
    d *= this.io / cost.getIo();
    ++n;
  }
  if (n == 0) {
    return 1.0;
  }
  return Math.pow(d, 1 / n);// 开1/n根号
}

public double computeSortIOCost(Pair<Double, Double> relationInfo) {
    //relationInfo：Pair类型<记录数，平均记录大小>
    // Sort-merge join
    double ioCost = 0.0;
    double cardinality = relationInfo.left;//基数
    double averageTupleSize = relationInfo.right;//平均元祖或记录大小
    // Write cost
    ioCost += cardinality * averageTupleSize * localFSWrite;
    // Read cost
    ioCost += cardinality * averageTupleSize * localFSRead;
    // Net transfer cost
    ioCost += cardinality * averageTupleSize * netCost;
    return ioCost; //返回总结果
  }

public double computeSortCPUCost(Double cardinality) {//传入参数为基数
  return cardinality * Math.log(cardinality) * cpuCost;
}

public double computeSortMergeIOCost(
        ImmutableList<Pair<Double, Double>> relationInfos) {//Pair类型<记录数，平均记录大小>
  // Sort-merge join
  double ioCost = 0.0;
  for (Pair<Double,Double> relationInfo : relationInfos) {  //列表的遍历
    ioCost += computeSortIOCost(relationInfo);//累加了Sort IO估算
  }
  return ioCost;
}

public double computeSortMergeCPUCost(
       ImmutableList<Double> cardinalities,
       ImmutableBitSet sorted) {
 // Sort-merge join
 double cpuCost = 0.0;
 for (int i=0; i<cardinalities.size(); i++) {
   double cardinality = cardinalities.get(i);
   if (!sorted.get(i)) {//BitSet位图判断是否存在
     // Sort cost
     // 排序CPU成本 = 基数 * 基数自然对数 * CPU成本
     cpuCost += computeSortCPUCost(cardinality); //累加单个CPU成本
  }
   // Merge cost
   //合并的成本 = 记录数 * CPU成本
   cpuCost += cardinality * cpuCost; //合并cpu成本
}
 return cpuCost;
}

public double computeMapJoinIOCost(
        ImmutableList<Pair<Double, Double>> relationInfos,////Pair类型<记录数，平均记录大小>
        ImmutableBitSet streaming, int parallelism) {
  // Hash-join
  double ioCost = 0.0;
  for (int i=0; i<relationInfos.size(); i++) {
    double cardinality = relationInfos.get(i).left; //获取基数大小
    double averageTupleSize = relationInfos.get(i).right;//平均记录大小
    if (!streaming.get(i)) {//判断为不在mapper内存中的
      ioCost += cardinality * averageTupleSize * netCost * parallelism;
    }
  }
  return ioCost;
}

public static double computeMapJoinCPUCost(
        ImmutableList<Double> cardinalities,
        ImmutableBitSet streaming) {
  // Hash-join
  double cpuCost = 0.0;
  for (int i=0; i<cardinalities.size(); i++) {
    double cardinality = cardinalities.get(i);
    if (!streaming.get(i)) {//判断进mapper内存中的
      cpuCost += cardinality;
    }
    cpuCost += cardinality * cpuCost;
  }
  return cpuCost;
}

public double computeBucketMapJoinIOCost(
        ImmutableList<Pair<Double, Double>> relationInfos,//Pair类型<记录数，平均记录大小>
        ImmutableBitSet streaming, int parallelism) {
  // Hash-join
  double ioCost = 0.0;
  for (int i=0; i<relationInfos.size(); i++) {
    double cardinality = relationInfos.get(i).left;
    double averageTupleSize = relationInfos.get(i).right;//平均记录（元组）大小
    if (!streaming.get(i)) {
      ioCost += cardinality * averageTupleSize * netCost * parallelism;
    }
  }
  return ioCost;
}

public double computeBucketMapJoinCPUCost(
        ImmutableList<Double> cardinalities,
        ImmutableBitSet streaming) {
  // Hash-join
  double cpuCost = 0.0;
  for (int i=0; i<cardinalities.size(); i++) {
    double cardinality = cardinalities.get(i);//基数
    if (!streaming.get(i)) {
      cpuCost += cardinality * cpuCost;
    }
    cpuCost += cardinality * cpuCost;
  }
  return cpuCost;
}

public double computeSMBMapJoinIOCost(
        ImmutableList<Pair<Double, Double>> relationInfos, //Pair类型<记录数，平均记录大小>
        ImmutableBitSet streaming, int parallelism) {
  // Hash-join
  double ioCost = 0.0;
  for (int i=0; i<relationInfos.size(); i++) {
    double cardinality = relationInfos.get(i).left;
    double averageTupleSize = relationInfos.get(i).right;
    if (!streaming.get(i)) {
      ioCost += cardinality * averageTupleSize * netCost * parallelism;
    }
  }
  return ioCost;
}

public static double computeSMBMapJoinCPUCost(
        ImmutableList<Double> cardinalities) { //基数列表
  // Hash-join
  double cpuCost = 0.0;
  for (int i=0; i<cardinalities.size(); i++) {
    cpuCost += cardinalities.get(i) * cpuCost;
  }
  return cpuCost;
}

  /***
 * 以下默认返回的都是0成本
 *
 * @return
 */
//默认成本HiveCost
@Override
public RelOptCost getDefaultCost() {
  return HiveCost.FACTORY.makeZeroCost();//new HiveCost(0.0, 0.0, 0.0)
}

//TableScan成本HiveCost
@Override
public RelOptCost getScanCost(HiveTableScan ts) {
  return HiveCost.FACTORY.makeZeroCost();//new HiveCost(0.0, 0.0, 0.0)
}

//Aggregate成本HiveCost
@Override
public RelOptCost getAggregateCost(HiveAggregate aggregate) {
  return HiveCost.FACTORY.makeZeroCost();//new HiveCost(0.0, 0.0, 0.0)
}

public RelOptCost getJoinCost(HiveJoin join) { //获取join成本，选取最小成本的算法。
  // Select algorithm with min cost
  JoinAlgorithm joinAlgorithm = null;
  RelOptCost minJoinCost = null;

  if (LOG.isTraceEnabled()) {
    LOG.trace("Join algorithm selection for:\n" + RelOptUtil.toString(join));
  }

  //遍历join算法集合，选取比较大小，选取最小的join成本
  for (JoinAlgorithm possibleAlgorithm : this.joinAlgorithms) {//遍历HiveCost集合，从Join算法中选择最小的成本作为返回值
    if (!possibleAlgorithm.isExecutable(join)) {
      continue;
    }
    RelOptCost joinCost = possibleAlgorithm.getCost(join);//获取初始化的第一个成本
    if (LOG.isTraceEnabled()) {
      LOG.trace(possibleAlgorithm + " cost: " + joinCost);
    }
    if (minJoinCost == null || joinCost.isLt(minJoinCost) ) {
      joinAlgorithm = possibleAlgorithm; //次最小成本
      minJoinCost = joinCost;
    }
  }
  if (LOG.isTraceEnabled()) {
    LOG.trace(joinAlgorithm + " selected");
  }
//当前成本最小的Join算法和成本大小
  join.setJoinAlgorithm(joinAlgorithm);
  join.setJoinCost(minJoinCost);

  return minJoinCost;
}

public static class DefaultJoinAlgorithm implements JoinAlgorithm {

  public static final JoinAlgorithm INSTANCE = new DefaultJoinAlgorithm();
  private static final String ALGORITHM_NAME = "none";

  @Override
  public String toString() {
    return ALGORITHM_NAME;
  }

  //默认可执行的
  @Override
  public boolean isExecutable(HiveJoin join) {
    return true;
  }
  //默认值只有行数（左右两侧记录数之和），内存，IO为0
  @Override
  public RelOptCost getCost(HiveJoin join) {
    RelMetadataQuery mq = RelMetadataQuery.instance();
    double leftRCount = mq.getRowCount(join.getLeft());//获取左侧记录数
    double rightRCount = mq.getRowCount(join.getRight());//获取右侧记录数
    return HiveCost.FACTORY.makeCost(leftRCount + rightRCount, 0.0, 0.0); //Creates a cost object
  }
  @Override
  public ImmutableList<RelCollation> getCollation(HiveJoin join) { //默认为空
    return ImmutableList.of();
  }

  //默认的分布式类型 单例
  @Override
  public RelDistribution getDistribution(HiveJoin join) {//默认值，单例
    return RelDistributions.SINGLETON; //物理分布类型，单例类型
  }
  @Override
  public Double getMemory(HiveJoin join) {//默认值，内存null为空
    return null;
  }

  @Override
  public Double getCumulativeMemoryWithinPhaseSplit(HiveJoin join) { //默认值，阶段性内存null为空
    return null;
  }
  @Override
  public Boolean isPhaseTransition(HiveJoin join) {//默认值，非事务
    return false;
  }
  //拆分个数默认为1个
  @Override
  public Integer getSplitCount(HiveJoin join) {//默认值，只有一个拆分
    return 1;
  }
}