Spark sql规则执行器RuleExecutor(源码解析)

abstract class RuleExecutor[TreeType <: TreeNode[_]] extends Logging {

abstract class Strategy {
  def maxIterations: Int
  def errorOnExceed: Boolean = false
  def maxIterationsSetting: String = null
}

case object Once extends Strategy { val maxIterations = 1 }

case class FixedPoint(
  override val maxIterations: Int,
  override val errorOnExceed: Boolean = false,
  override val maxIterationsSetting: String = null) extends Strategy

protected case class Batch(name: String, strategy: Strategy, rules: Rule[TreeType]*)

protected def batches: Seq[Batch]

abstract class Strategy {

  //表示 Batch 最大执行次数,有可能是一次或多次。如果执行了 maxIterations 次之前达到收敛点,就停止，不再继续执行 Batch
  def maxIterations: Int

  //决定如果执行次超过了最大迭代次数是否抛出异常
  def errorOnExceed: Boolean = false

  //是大执行次数可以在配置文件里配制，这里是获取配制文件中key。
  def maxIterationsSetting: String = null
}

case object Once extends Strategy { val maxIterations = 1 }

case class FixedPoint(
  override val maxIterations: Int,
  override val errorOnExceed: Boolean = false,
  override val maxIterationsSetting: String = null) extends Strategy

/** A batch of rules. */
protected case class Batch(
name: String, //这组同类规则的名称
strategy: Strategy,//策略
rules: Rule[TreeType]* //具体的规则 
)
 

//传入参数plan是当前的执行计划
def execute(plan: TreeType): TreeType = {
  var curPlan = plan
  val queryExecutionMetrics = RuleExecutor.queryExecutionMeter
  val planChangeLogger = new PlanChangeLogger()
  val tracker: Option[QueryPlanningTracker] = QueryPlanningTracker.get
  val beforeMetrics = RuleExecutor.getCurrentMetrics()

  // Run the structural integrity checker against the initial input
  if (!isPlanIntegral(plan)) {
    val message = "The structural integrity of the input plan is broken in " +
      s"${this.getClass.getName.stripSuffix("$")}."
    throw new TreeNodeException(plan, message, null)
  }

  //遍历batches，取出batch
  batches.foreach { batch =>
    //针对每个batch处理
    val batchStartPlan = curPlan
    var iteration = 1 
    var lastPlan = curPlan
    var continue = true
    while (continue) {
      //foldLeft语法，形如A.foldLeft(B){(z,i)=>{dosomething() return C}}
      //curPlan是batch迭代的初始值
      //case(plan,rule) 中plan是batch中某个rule执行后返回的plan，继续投入batch中下个rule；rule是指batch中的某个rule对象
      //整个foldLeft最后返回当前batch执行完之后生成的plan
      curPlan = batch.rules.foldLeft(curPlan) {
        case (plan, rule) =>
          val startTime = System.nanoTime()
          val result = rule(plan)  //调用rule的apply方法，执行规则       
          val runTime = System.nanoTime() - startTime
          val effective = !result.fastEquals(plan)

          if (effective) {
            queryExecutionMetrics.incNumEffectiveExecution(rule.ruleName)
            queryExecutionMetrics.incTimeEffectiveExecutionBy(rule.ruleName, runTime)
            planChangeLogger.logRule(rule.ruleName, plan, result)
          }
          queryExecutionMetrics.incExecutionTimeBy(rule.ruleName, runTime)
          queryExecutionMetrics.incNumExecution(rule.ruleName)

          // Record timing information using QueryPlanningTracker
          tracker.foreach(_.recordRuleInvocation(rule.ruleName, runTime, effective))

          // Run the structural integrity checker against the plan after each rule.
          if (!isPlanIntegral(result)) {
            val message = s"After applying rule ${rule.ruleName} in batch ${batch.name}, " +
              "the structural integrity of the plan is broken."
            throw new TreeNodeException(result, message, null)
          }

          result
      }
      iteration += 1 //迭代次数+1
      if (iteration > batch.strategy.maxIterations) {
         //处理迭代次数大于策略中最大迭代次的情况
        
        if (iteration != 2) {
         //这个处理策略最大迭代次数不是一次，并且迭代次数超过了maxIterations的情况
          val endingMsg = if (batch.strategy.maxIterationsSetting == null) {
            "."
          } else {
            s", please set '${batch.strategy.maxIterationsSetting}' to a larger value."
          }
          val message = s"Max iterations (${iteration - 1}) reached for batch ${batch.name}" +
            s"$endingMsg"
          if (Utils.isTesting || batch.strategy.errorOnExceed) {
           //如果errorOnExceed为true则抛出异常 
            throw new TreeNodeException(curPlan, message, null)
          } else {
            //只给警告日志
            logWarning(message)
          }
        }
        // Check idempotence for Once batches.
        if (batch.strategy == Once &&
          Utils.isTesting && !blacklistedOnceBatches.contains(batch.name)) {
          checkBatchIdempotence(batch, curPlan)
        }
        //对于达到策略设置数量，continue置false，迭代将会结束
        continue = false
      }
    //如果迭代次数没达到maxIterations，但是当前plan和上次plan完全一样的话，也会把 continue 置为 false，停止迭代
      if (curPlan.fastEquals(lastPlan)) {
        logTrace(
          s"Fixed point reached for batch ${batch.name} after ${iteration - 1} iterations.")
        continue = false
      }
      lastPlan = curPlan
    }

    planChangeLogger.logBatch(batch.name, batchStartPlan, curPlan)
  }
  planChangeLogger.logMetrics(RuleExecutor.getCurrentMetrics() - beforeMetrics)

  curPlan
}