class Analyzer(catalog: Catalog, registry: FunctionRegistry, caseSensitive: Boolean)  
  extends RuleExecutor[LogicalPlan] with HiveTypeCoercion {  
  
  // TODO: pass this in as a parameter.  
  val fixedPoint = FixedPoint(100)  
  
  val batches: Seq[Batch] = Seq(  
    Batch("MultiInstanceRelations", Once,  
      NewRelationInstances),  
    Batch("CaseInsensitiveAttributeReferences", Once,  
      (if (caseSensitive) Nil else LowercaseAttributeReferences :: Nil) : _*),  
    Batch("Resolution", fixedPoint,  
      ResolveReferences ::  
      ResolveRelations ::  
      NewRelationInstances ::  
      ImplicitGenerate ::  
      StarExpansion ::  
      ResolveFunctions ::  
      GlobalAggregates ::  
      typeCoercionRules :_*),  
    Batch("AnalysisOperators", fixedPoint,  
      EliminateAnalysisOperators)  
  )  

/** A strategy that runs until fix point or maxIterations times, whichever comes first. */  
case class FixedPoint(maxIterations: Int) extends Strategy  

/** A batch of rules. */£¬  
protected case class Batch(name: String, strategy: Strategy, rules: Rule[TreeType]*) 

abstract class Rule[TreeType <: TreeNode[_]] extends Logging {  
  
  /** Name for this rule, automatically inferred based on class name. */  
  val ruleName: String = {  
    val className = getClass.getName  
    if (className endsWith "$") className.dropRight(1) else className  
  }  
  
  def apply(plan: TreeType): TreeType  
}  

/** 
 * An execution strategy for rules that indicates the maximum number of executions. If the 
 * execution reaches fix point (i.e. converge) before maxIterations, it will stop. 
 */  
abstract class Strategy { def maxIterations: Int }  

def apply(plan: TreeType): TreeType = {  
   var curPlan = 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) {  
       curPlan = batch.rules.foldLeft(curPlan) {  
         case (plan, rule) =>  
           val result = rule(plan) 
//ÕâÀｫµ÷Óø÷¸ö²»Í¬RuleµÄapply·½·¨£¬½«UnResolved Relations£¬AttrubuteºÍFunction½øÐÐResolve  
           if (!result.fastEquals(plan)) {  
             logger.trace(  
               s"""  
                 |=== Applying Rule ${rule.ruleName} ===  
                 |${sideBySide(plan.treeString, result.treeString).mkString("\n")}  
               """.stripMargin)  
           }  
  
           result //·µ»Ø×÷ÓúóµÄresult plan  
       }  
       iteration += 1  
       if (iteration > batch.strategy.maxIterations) { //Èç¹ûµü´ú´ÎÊýÒѾ­´óÓڸòßÂÔµÄ×î´óµü´ú´ÎÊý£¬¾Íֹͣѭ»·  
         logger.info(s"Max iterations ($iteration) reached for batch ${batch.name}")  
         continue = false  
       }  
  
       if (curPlan.fastEquals(lastPlan)) { //Èç¹ûÔÚ¶à´Îµü´úÖв»Ôٱ仯£¬ÒòΪplanÓиöunique id£¬¾Íֹͣѭ»·¡£  
         logger.trace(s"Fixed point reached for batch ${batch.name} after $iteration iterations.")  
         continue = false  
       }  
       lastPlan = curPlan  
     }  
  
     if (!batchStartPlan.fastEquals(curPlan)) {  
       logger.debug(  
         s"""  
         |=== Result of Batch ${batch.name} ===  
         |${sideBySide(plan.treeString, curPlan.treeString).mkString("\n")}  
       """.stripMargin)  
     } else {  
       logger.trace(s"Batch ${batch.name} has no effect.")  
     }  
   }  
  
   curPlan //·µ»ØResolvedµÄLogical Plan  
 } 

Batch("MultiInstanceRelations", Once,  
     NewRelationInstances)  

trait MultiInstanceRelation {  
  def newInstance: this.type  
}  

object NewRelationInstances extends Rule[LogicalPlan] {   
  def apply(plan: LogicalPlan): LogicalPlan = {  
    val localRelations = plan collect { case l: MultiInstanceRelation => l}
 //½«logical planÓ¦ÓÃpartial functionµÃµ½ËùÓÐMultiInstanceRelationµÄplanµÄ¼¯ºÏ   
    val multiAppearance = localRelations  
      .groupBy(identity[MultiInstanceRelation]) //group by²Ù×÷  
      .filter { case (_, ls) => ls.size > 1 } //Èç¹ûֻȡsize´óÓÚ1µÄ½øÐкóÐø²Ù×÷  
      .map(_._1)  
      .toSet  
  
    //¸üÐÂplan£¬Ê¹µÃÿ¸öʵÀýµÄexpIdÊÇΨһµÄ¡£  
    plan transform {  
      case l: MultiInstanceRelation if multiAppearance contains l => l.newInstance  
    }  
  }  
}  

object LowercaseAttributeReferences extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case UnresolvedRelation(databaseName, name, alias) =>  
      UnresolvedRelation(databaseName, name, alias.map(_.toLowerCase))  
    case Subquery(alias, child) => Subquery(alias.toLowerCase, child)  
    case q: LogicalPlan => q transformExpressions {  
      case s: Star => s.copy(table = s.table.map(_.toLowerCase))  
      case UnresolvedAttribute(name) => UnresolvedAttribute(name.toLowerCase)  
      case Alias(c, name) => Alias(c, name.toLowerCase)()  
      case GetField(c, name) => GetField(c, name.toLowerCase)  
    }  
  }  
}  

object ResolveReferences extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {  
    case q: LogicalPlan if q.childrenResolved =>  
      logger.trace(s"Attempting to resolve ${q.simpleString}")  
      q transformExpressions {  
        case u @ UnresolvedAttribute(name) =>  
          // Leave unchanged if resolution fails.  Hopefully will be resolved next round.  
          val result = q.resolve(name).getOrElse(u)//ת»¯ÎªNamedExpression  
          logger.debug(s"Resolving $u to $result")  
          result  
      }  
  }  
}  

object ResolveRelations extends Rule[LogicalPlan] {  
    def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
      case UnresolvedRelation(databaseName, name, alias) =>  
        catalog.lookupRelation(databaseName, name, alias)  
    }  
  } 

object ImplicitGenerate extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case Project(Seq(Alias(g: Generator, _)), child) =>  
      Generate(g, join = false, outer = false, None, child)  
  }  
}  

object StarExpansion extends Rule[LogicalPlan] {  
    def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
      // Wait until children are resolved  
      case p: LogicalPlan if !p.childrenResolved => p  
      // If the projection list contains Stars, expand it.  
      case p @ Project(projectList, child) if containsStar(projectList) =>   
        Project(  
          projectList.flatMap {  
            case s: Star => s.expand(child.output) 
//Õ¹¿ª£¬½«ÊäÈëµÄAttributeexpand(input: Seq[Attribute]) ת»¯ÎªSeq[NamedExpression]  
            case o => o :: Nil  
          },  
          child)  
      case t: ScriptTransformation if containsStar(t.input) =>  
        t.copy(  
          input = t.input.flatMap {  
            case s: Star => s.expand(t.child.output)  
            case o => o :: Nil  
          }  
        )  
      // If the aggregate function argument contains Stars, expand it.  
      case a: Aggregate if containsStar(a.aggregateExpressions) =>  
        a.copy(  
          aggregateExpressions = a.aggregateExpressions.flatMap {  
            case s: Star => s.expand(a.child.output)  
            case o => o :: Nil  
          }  
        )  
    }  
    /** 
     * Returns true if `exprs` contains a [[Star]]. 
     */  
    protected def containsStar(exprs: Seq[Expression]): Boolean =  
      exprs.collect { case _: Star => true }.nonEmpty  
  }  
} 

object ResolveFunctions extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case q: LogicalPlan =>  
      q transformExpressions {  
        case u @ UnresolvedFunction(name, children) if u.childrenResolved =>  
          registry.lookupFunction(name, children) //¿´ÊÇ·ñ×¢²áÁ˵±Ç°udf  
      }  
  }  
}  

object GlobalAggregates extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case Project(projectList, child) if containsAggregates(projectList) =>  
      Aggregate(Nil, projectList, child)  
  }  
  
  def containsAggregates(exprs: Seq[Expression]): Boolean = {  
    exprs.foreach(_.foreach {  
      case agg: AggregateExpression => return true  
      case _ =>  
    })  
    false  
  }  
}  

val typeCoercionRules =  
  PropagateTypes ::  
  ConvertNaNs ::  
  WidenTypes ::  
  PromoteStrings ::  
  BooleanComparisons ::  
  BooleanCasts ::  
  StringToIntegralCasts ::  
  FunctionArgumentConversion ::  
  CastNulls ::  
  Nil  

object EliminateAnalysisOperators extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {  
    case Subquery(_, child) => child //Óöµ½Subquery,²»·´»Ú±¾Éí£¬·µ»ØËüµÄChild£¬¼´É¾³ýÁ˸ÃÔªËØ  
    case LowerCaseSchema(child) => child  
  }  
}  

val exec = sqlContext.sql("select mobile as mb, sid as id, mobile*2 multi2mobile,
 count(1) times from (select * from temp_shengli_mobile)a where pfrom_id=0.0 group by mobile, sid,  mobile*2")  
14/07/21 18:23:32 DEBUG SparkILoop$SparkILoopInterpreter: Invoking: public static java.lang.String $line47.$eval.$print()  
14/07/21 18:23:33 INFO Analyzer: Max iterations (2) reached for batch MultiInstanceRelations  
14/07/21 18:23:33 INFO Analyzer: Max iterations (2) reached for batch CaseInsensitiveAttributeReferences  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'pfrom_id to pfrom_id#5  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'mobile to mobile#2  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'sid to sid#1  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'mobile to mobile#2  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'mobile to mobile#2  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'sid to sid#1  
14/07/21 18:23:33 DEBUG Analyzer$ResolveReferences$: Resolving 'mobile to mobile#2  
14/07/21 18:23:33 DEBUG Analyzer:   
=== Result of Batch Resolution ===  
!Aggregate ['mobile,'sid,('mobile * 2) AS c2#27], 
['mobile AS mb#23,'sid AS id#24,('mobile * 2) AS multi2mobile#25,COUNT(1) AS times#26L]  
Aggregate [mobile#2,sid#1,(CAST(mobile#2, DoubleType) * CAST(2, DoubleType)) AS c2#27],
[mobile#2 AS mb#23,sid#1 AS id#24,(CAST(mobile#2, DoubleType) * CAST(2, DoubleType)) 
AS multi2mobile#25,COUNT(1) AS times#26L]  
! Filter ('pfrom_id = 0.0)   
 Filter (CAST(pfrom_id#5, DoubleType) = 0.0)  
   Subquery a    
Subquery a  
!   Project [*] 
 Project [data_date#0,sid#1,mobile#2,pverify_type#3,create_time#4,
pfrom_id#5,p_status#6,pvalidate_time#7,feffect_time#8,plastupdate_ip#9,update_time#10,status#11,preserve_int#12]  
!    UnresolvedRelation None, temp_shengli_mobile, None 
Subquery temp_shengli_mobile  
!     SparkLogicalPlan (ExistingRdd [data_date#0,sid#1,mobile#2,pverify_type#3,create_time#4,
pfrom_id#5,p_status#6,pvalidate_time#7,feffect_time#8,plastupdate_ip#9,update_time#10,status#11,
preserve_int#12], MapPartitionsRDD[4] at mapPartitions at basicOperators.scala:174)  
14/07/21 18:23:33 DEBUG Analyzer:   
=== Result of Batch AnalysisOperators ===  
!Aggregate ['mobile,'sid,('mobile * 2) AS c2#27], ['mobile AS mb#23,'sid AS id#24,('mobile * 2) 
AS multi2mobile#25,COUNT(1) AS times#26L]  
 Aggregate [mobile#2,sid#1,(CAST(mobile#2, DoubleType) * CAST(2, DoubleType)) AS c2#27],
 [mobile#2 AS mb#23,sid#1 AS id#24,(CAST(mobile#2, DoubleType) * CAST(2, DoubleType))
 AS multi2mobile#25,COUNT(1) AS times#26L]  
! Filter ('pfrom_id = 0.0) 
 Filter (CAST(pfrom_id#5, DoubleType) = 0.0)  
!  Subquery a  
 Project [data_date#0,sid#1,mobile#2,pverify_type#3,create_time#4,
pfrom_id#5,p_status#6,pvalidate_time#7,feffect_time#8,plastupdate_ip#9,update_time#10,status#11,preserve_int#12]  
!   Project [*]   
SparkLogicalPlan (ExistingRdd [data_date#0,sid#1,mobile#2,pverify_type#3,
create_time#4,pfrom_id#5,p_status#6,pvalidate_time#7,feffect_time#8,plastupdate_ip#9,
update_time#10,status#11,preserve_int#12], MapPartitionsRDD[4] at mapPartitions at basicOperators.scala:174)  
!    UnresolvedRelation None, temp_shengli_mobile, None