Spark源码系列（九）Spark SQL初体验之解析过程详解

  // ~>只保留右边的模式 opt可选的 ~按顺序合成 <~只保留左边的
  protected lazy val select: Parser[LogicalPlan] =
    SELECT ~> opt(DISTINCT) ~ projections ~
    opt(from) ~ opt(filter) ~
    opt(grouping) ~
    opt(having) ~
    opt(orderBy) ~
    opt(limit) <~ opt(";") ^^ {
      case d ~ p ~ r ~ f ~ g ~ h ~ o ~ l  =>
        val base = r.getOrElse(NoRelation)
        val withFilter = f.map(f => Filter(f, base)).getOrElse(base)
        val withProjection =
          g.map {g =>
            Aggregate(assignAliases(g), assignAliases(p), withFilter)
          }.getOrElse(Project(assignAliases(p), withFilter))
        val withDistinct = d.map(_ => Distinct(withProjection)).getOrElse(withProjection)
        val withHaving = h.map(h => Filter(h, withDistinct)).getOrElse(withDistinct)
        val withOrder = o.map(o => Sort(o, withHaving)).getOrElse(withHaving)
        val withLimit = l.map { l => Limit(l, withOrder) }.getOrElse(withOrder)
        withLimit
  }

  def apply(plan: TreeType): TreeType = {
    var curPlan = plan
    //规则还分批次的，分批对plan进行处理
    batches.foreach { batch =>
      val batchStartPlan = curPlan
      var iteration = 1
      var lastPlan = curPlan
      var continue = true

      // Run until fix point (or the max number of iterations as specified in the strategy.
      while (continue) {
        //用batch种的小规则从左到右挨个对plan进行处理
        curPlan = batch.rules.foldLeft(curPlan) {
          case (plan, rule) =>
            val result = rule(plan)
            result
        }
        iteration += 1
        //超过了规定的迭代次数就要退出的
        if (iteration > batch.strategy.maxIterations) {
              continue = false
        }
        //经过处理成功的plan是会发生改变的，如果和上一次处理接触的plan一样，这说明已经没有优化空间了，可以结束，这个就是前面提到的Fixed point
        if (curPlan.fastEquals(lastPlan)) {
          continue = false
        }
        lastPlan = curPlan
      }
    }

    curPlan
  }

  val batches: Seq[Batch] = Seq(
    //碰到继承自MultiInstanceRelations接口的LogicPlan时，发现id以后重复的，就强制要求它们生成一个新的全局唯一的id
    //涉及到InMemoryRelation、LogicRegion、ParquetRelation、SparkLogicPlan
    Batch("MultiInstanceRelations", Once,
      NewRelationInstances),
    //如果大小写不敏感就把属性都变成小写
    Batch("CaseInsensitiveAttributeReferences", Once,
      (if (caseSensitive) Nil else LowercaseAttributeReferences :: Nil) : _*),
    //这个牛逼啊，居然想迭代100次的。
    Batch("Resolution", fixedPoint,
      //解析从子节点的操作生成的属性，一般是别名引起的，比如a.id
      ResolveReferences ::
      //通过catalog解析表名
      ResolveRelations ::
      //在select语言里，order by的属性往往在前面没写，查询的时候也需要把这些字段查出来，排序完毕之后再删除
      ResolveSortReferences ::
      //前面讲过了
      NewRelationInstances ::
      //清除被误认为别名的属性，比如sum(score) as a，其实它应该是sum(score)才对
      //它被解析的时候解析成Project(Seq(Alias(g: Generator, _))，直接返回Generator就可以了
      ImplicitGenerate ::
      //处理语句中的*，比如select *, count(*)
      StarExpansion ::
      //解析函数
      ResolveFunctions ::
      //解析全局的聚合函数，比如select sum(score) from table
      GlobalAggregates ::
      //解析having子句后面的聚合过滤条件，比如having sum(score) > 400
      UnresolvedHavingClauseAttributes ::
      //typeCoercionRules是hive的类型转换规则
      typeCoercionRules :_*),
    //检查所有节点的属性是否都已经处理完毕了，如果还有没解析出来的属性，这里就会报错！
    Batch("Check Analysis", Once,
      CheckResolution),
    //清除多余的操作符，现在是Subquery和LowerCaseSchema，
    //第一个是子查询，第二个HiveContext查询树里面把子节点全部转换成小写
    Batch("AnalysisOperators", fixedPoint,
      EliminateAnalysisOperators)
  )

object Optimizer extends RuleExecutor[LogicalPlan] {
  val batches =
      //递归合并相邻的两个limit
    Batch("Combine Limits", FixedPoint(100),
      CombineLimits) ::
    Batch("ConstantFolding", FixedPoint(100),
      //替换null值
      NullPropagation,
      //替换一些简单的常量表达式，比如 1 in (1,2) 直接返回一个true就可以了
      ConstantFolding,
      //简化like语句，避免全表扫描，目前支持‘%demo%‘, ‘%demo‘,‘demo*‘,‘demo‘
      LikeSimplification,
      //简化过滤条件，比如true and score > 0 直接替换成score > 0
      BooleanSimplification,
      //简化filter，比如where 1=1 或者where 1=2，前者直接去掉这个过滤，后者这个查询就没必要做了
      SimplifyFilters,
      //简化转换，比如两个比较字段的数据类型是一样的，就不需要转换了
      SimplifyCasts,
      //简化大小写转换，比如Upper(Upper(‘a‘))转为认为是Upper(‘a‘)
      SimplifyCaseConversionExpressions) ::
    Batch("Filter Pushdown", FixedPoint(100),
      //递归合并相邻的两个过滤条件
      CombineFilters,
      //把从表达式里面的过滤替换成，先做过滤再取表达式，并且掉过滤里面的别名属性
      //典型的例子 select * from (select a,b from table) where a=1
      //替换成select * from (select a,b from table where a=1)
      PushPredicateThroughProject,
      //把join的on条件中可以在原表当中做过滤的先做过滤
      //比如select a,b from x join y on x.id = y.id and x.a >0 and y.b >0
      //这个语句可以改写为 select a,b from x where x.a > 0 join (select * from y where y.b >0) on x.id = y.id
      PushPredicateThroughJoin,
      //去掉一些用不上的列
      ColumnPruning) :: Nil
}

  protected[sql] class SparkPlanner extends SparkStrategies {
    val sparkContext: SparkContext = self.sparkContext

    val sqlContext: SQLContext = self

    def codegenEnabled = self.codegenEnabled

    def numPartitions = self.numShufflePartitions
    //把LogicPlan转换成实际的操作，具体操作类在org.apache.spark.sql.execution包下面
    val strategies: Seq[Strategy] =
      //把cache、set、expain命令转化为实际的Command
      CommandStrategy(self) ::
      //把limit转换成TakeOrdered操作
      TakeOrdered ::
      //名字有点蛊惑人，就是转换聚合操作
      HashAggregation ::
      //left semi join只显示连接条件成立的时候连接左边的表的信息
      //比如select * from table1 left semi join table2 on(table1.student_no=table2.student_no);
      //它只显示table1中student_no在表二当中的信息，它可以用来替换exist语句
      LeftSemiJoin ::
      //等值连接操作，有些优化的内容，如果表的大小小于spark.sql.autoBroadcastJoinThreshold设置的字节
      //就自动转换为BroadcastHashJoin，即把表缓存，类似hive的map join（顺序是先判断右表再判断右表）。
      //这个参数的默认值是10000
      //另外做内连接的时候还会判断左表右表的大小，shuffle取数据大表不动，从小表拉取数据过来计算
      HashJoin ::
      //在内存里面执行select语句进行过滤,会做缓存
      InMemoryScans ::
      //和parquet相关的操作
      ParquetOperations ::
      //基本的操作
      BasicOperators ::
      //没有条件的连接或者内连接做笛卡尔积
      CartesianProduct ::
      //把NestedLoop连接进行广播连接
      BroadcastNestedLoopJoin :: Nil
      ......  
}