SELECT  a1,a2,a3  FROM  tableA  Where  condition 

• TreeNode可以细分成三种类型的Node：
  • UnaryNode 一元节点，即只有一个子节点。如Limit、Filter操作
  • BinaryNode 二元节点，即有左右子节点的二叉节点。如Jion、Union操作
  • LeafNode 叶子节点，没有子节点的节点。主要用户命令类操作，如SetCommand

• Rule的相关代码定义在sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules
• Rule在sparkSQL的Analyzer、Optimizer、SparkPlan等各个组件中都有应用到
• Rule是一个抽象类，具体的Rule实现是通过RuleExecutor完成
• Rule通过定义batch和batchs，可以简便的、模块化地对Tree进行transform操作
• Rule通过定义Once和FixedPoint，可以对Tree进行一次操作或多次操作（如对某些Tree进行多次迭代操作的时候，达到FixedPoint次数迭代或达到前后两次的树结构没变化才停止操作，具体参看RuleExecutor.apply）

/**源自sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala  */
  def sql(sqlText: String): SchemaRDD = {
    if (dialect == "sql") {
      new SchemaRDD(this, parseSql(sqlText))   //parseSql(sqlText)对sql语句进行语法解析
    } else {
      sys.error(s"Unsupported SQL dialect: $dialect")
    }
  }

/**源自sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala  */
  protected[sql] val parser = new catalyst.SqlParser
  protected[sql] def parseSql(sql: String): LogicalPlan = parser(sql)

/**源自sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala  */
class SchemaRDD(
    @transient val sqlContext: SQLContext,
    @transient val baseLogicalPlan: LogicalPlan)
  extends RDD[Row](sqlContext.sparkContext, Nil) with SchemaRDDLike

/**源自sql/core/src/main/scala/org/apache/spark/sql/SchemaRDDLike.scala  */
private[sql] trait SchemaRDDLike {
  @transient val sqlContext: SQLContext
  @transient val baseLogicalPlan: LogicalPlan
  private[sql] def baseSchemaRDD: SchemaRDD

  lazy val queryExecution = sqlContext.executePlan(baseLogicalPlan)

/**源自sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala  */
protected abstract class QueryExecution {
    def logical: LogicalPlan
   
    //对Unresolved LogicalPlan进行analyzer，生成resolved LogicalPlan
    lazy val analyzed = ExtractPythonUdfs(analyzer(logical))    
    //对resolved LogicalPlan进行optimizer，生成optimized LogicalPlan
    lazy val optimizedPlan = optimizer(analyzed)  
    // 将optimized LogicalPlan转换成PhysicalPlan
    lazy val sparkPlan = {
      SparkPlan.currentContext.set(self)
      planner(optimizedPlan).next()
    }
    // PhysicalPlan执行前的准备工作，生成可执行的物理计划
    lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)

    //执行可执行物理计划
    lazy val toRdd: RDD[Row] = executedPlan.execute()

    ......
  }

1. SQL语句经过SqlParse解析成UnresolvedLogicalPlan；
2. 使用analyzer结合数据数据字典（catalog）进行绑定，生成resolvedLogicalPlan；
3. 使用optimizer对resolvedLogicalPlan进行优化，生成optimizedLogicalPlan；
4. 使用SparkPlan将LogicalPlan转换成PhysicalPlan；
5. 使用prepareForExecution()将PhysicalPlan转换成可执行物理计划；
6. 使用execute()执行可执行物理计划；
7. 生成SchemaRDD。

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */
override def sql(sqlText: String): SchemaRDD = {
    // 使用spark.sql.dialect定义采用的语法解析器
    if (dialect == "sql") {
      super.sql(sqlText)    //如果使用sql解析器，则使用sqlContext的sql方法
    } else if (dialect == "hiveql") {     //如果使用和hiveql解析器，则使用HiveQl.parseSql
      new SchemaRDD(this, HiveQl.parseSql(sqlText))
    }  else {
      sys.error(s"Unsupported SQL dialect: $dialect.  Try 'sql' or 'hiveql'")
    }
  }

/**源自src/main/scala/org/apache/spark/sql/hive/HiveQl.scala  */
  /** Returns a LogicalPlan for a given HiveQL string. */
  def parseSql(sql: String): LogicalPlan = {
    try {
      if (条件)   {
      //非hive命令的处理，如set、cache table、add jar等直接转化成command类型的LogicalPlan
      .....  
      } else {
        val tree = getAst(sql)
        if (nativeCommands contains tree.getText) {
          NativeCommand(sql)
        } else {
          nodeToPlan(tree) match {
            case NativePlaceholder => NativeCommand(sql)
            case other => other
          }
        }
      }
    } catch {
      //异常处理
      ......
    }
  }

/**源自src/main/scala/org/apache/spark/sql/hive/HiveQl.scala  */  
/** * Returns the AST for the given SQL string.    */
  def getAst(sql: String): ASTNode = ParseUtils.findRootNonNullToken((new ParseDriver).parse(sql))

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */ 
override protected[sql] def executePlan(plan: LogicalPlan): this.QueryExecution =
    new this.QueryExecution { val logical = plan }

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */
 protected[sql] abstract class QueryExecution extends super.QueryExecution {
    // TODO: Create mixin for the analyzer instead of overriding things here.
    override lazy val optimizedPlan =
      optimizer(ExtractPythonUdfs(catalog.PreInsertionCasts(catalog.CreateTables(analyzed))))

    override lazy val toRdd: RDD[Row] = executedPlan.execute().map(_.copy())
    ......
  }

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */
/* A catalyst metadata catalog that points to the Hive Metastore. */
  @transient
  override protected[sql] lazy val catalog = new HiveMetastoreCatalog(this) with OverrideCatalog {
    override def lookupRelation(
      databaseName: Option[String],
      tableName: String,
      alias: Option[String] = None): LogicalPlan = {

      LowerCaseSchema(super.lookupRelation(databaseName, tableName, alias))
    }
  }

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */
  /* An analyzer that uses the Hive metastore. */
  @transient
  override protected[sql] lazy val analyzer =
    new Analyzer(catalog, functionRegistry, caseSensitive = false)

/**源自sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala  */
  @transient
  override protected[sql] val planner = hivePlanner

1. SQL语句经过HiveQl.parseSql解析成Unresolved LogicalPlan，在这个解析过程中对hiveql语句使用getAst()获取AST树，然后再进行解析；
2. 使用analyzer结合数据hive源数据Metastore（新的catalog）进行绑定，生成resolved LogicalPlan；
3. 使用optimizer对resolved LogicalPlan进行优化，生成optimized LogicalPlan，优化前使用了ExtractPythonUdfs(catalog.PreInsertionCasts(catalog.CreateTables(analyzed)))进行预处理；
4. 使用hivePlanner将LogicalPlan转换成PhysicalPlan；
5. 使用prepareForExecution()将PhysicalPlan转换成可执行物理计划；
6. 使用execute()执行可执行物理计划；
7. 执行后，使用map(_.copy)将结果导入SchemaRDD。