Spark SQL逻辑计划原理

Catalyst优化器是Spark引擎中非常重要的组成部分，也是近年来Spark社区项目重点投入、并且发展十分迅速的核心模块，对于Spark任务的性能提升起到了关键的基础作用。

我们知道，在Spark1.6之前开发人员是通过Spark的RDD编程接口来实现对大规模数据的分析和处理的，到了Spark1.6版本后推出了DataSet和DataFrame的编程接口，这种数据结构与RDD的主要区别在于其携带了结构化数据的Schema信息，从而可以被Spark Catalyst用来做进一步的解析和优化；而Spark SQL则是比DataSet和DataFrame编程接口更为简单易用的大数据领域语言，其用户可以是开发工程师、数据科学家、数据分析师等，并且与其他SQL语言类似，可以通过SQL引擎将SQL预先解析成一棵AST抽象语法树；同时，AST抽象语法树、DataSet及DataFrame接下来均会被Spark Catalyst优化器转换成为Unresolved LogicalPlan、Resolved LogicalPlan，Physical Plan、以及Optimized PhysicalPlan，也就是说带有schema信息的Spark分布式数据集都可以从Spark Catalyst中受益，这也是Spark任务性能得以提升的核心所在。

值得一提的是，在物理计划树的生成过程中，首先会将数据源解析成为RDD，也即在Spark SQL的物理计划执行过程中所操作的对象实际是RDD，一条Spark SQL在生成最终的物理计划后仍然会经过前面文章中所提到的生成DAG、划分Stage、并将taskset分发到特定的executor上运行等一系列的任务调度和执行过程来实现该Spark SQL的处理逻辑。

接下来，本文将着重讲解Spark SQL逻辑计划的相关实现原理，在后续的文章中会继续解析Spark SQL的物理计划。

生成Unresolved LogicalPlan

用户可以通过spark-sql等客户端来提交sql语句，在sparksession初始化时通过BaseSessionStateBuilder的build()方法始化SparkSqlParser、Analyser以及SparkOptimizer对象等：

def build(): SessionState = {
  new SessionState(
    session.sharedState,
    conf,
    experimentalMethods,
    functionRegistry,
    udfRegistration,
    () => catalog,
    sqlParser,
    () => analyzer,
    () => optimizer,
    planner,
    () => streamingQueryManager,
    listenerManager,
    () => resourceLoader,
    createQueryExecution,
    createClone,
    columnarRules,
    queryStagePrepRules)
}

当用户程序调用SparkSession的sql接口时即开始了解析sql语句并执行对数据处理的过程：

def sql(sqlText: String): DataFrame = withActive {
  val tracker = new QueryPlanningTracker
  val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
    sessionState.sqlParser.parsePlan(sqlText)
  }
  Dataset.ofRows(self, plan, tracker)
}

其中通过AbstractSqlParser的parsePlan方法将sql语句转换成抽象语法树：

override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
  astBuilder.visitSingleStatement(parser.singleStatement()) match {
    case plan: LogicalPlan => plan
    case _ =>
      val position = Origin(None, None)
      throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position)
  }
}

1、从SqlBaseParser的singleStatement()方法开始基于ANTLR4 lib库来解析sql语句中所有的词法片段，生成一棵AST抽象语法树；

2、访问AST抽象语法树并生成Unresolved 逻辑计划树：

1）访问SingleStatementContext节点：

SingleStatementContext是整个抽象语法树的根节点，因此以AstBuilder的visitSingleStatement方法为入口来遍历抽象语法树：

override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan = withOrigin(ctx) {
  visit(ctx.statement).asInstanceOf[LogicalPlan]
}
...
public T visit(ParseTree tree) {
   return tree.accept(this);
}

2）根据访问者模式执行SingleStatementContext节点的accept方法：

@Override
public <T> T accept(ParseTreeVisitor<? extends T> visitor) {
   if ( visitor instanceof SqlBaseVisitor ) return ((SqlBaseVisitor<? extends T>)visitor).visitSingleStatement(this);
   else return visitor.visitChildren(this);
}
...
@Override public T visitSingleStatement(SqlBaseParser.SingleStatementContext ctx) { return visitChildren(ctx); }

3）迭代遍历整棵AST Tree：

@Override
public T visitChildren(RuleNode node) {
   T result = defaultResult();
   int n = node.getChildCount();
   for (int i=0; i<n; i++) {
      if (!shouldVisitNextChild(node, result)) {
         break;
      }

      ParseTree c = node.getChild(i);
      T childResult = c.accept(this);
      result = aggregateResult(result, childResult);
   }

   return result;
}

根据以上代码，在遍历AST 树的过程中，会首先解析父节点的所有子节点，并执行子节点上的accept方法来进行解析，当所有子节点均解析为UnresolvedRelation或者Expression后，将这些结果进行聚合并返回到父节点，由此可见，AST树的遍历所采用的是后序遍历模式。

接下来以查询语句中的QuerySpecificationContext节点的解析为例进一步阐述以上过程：

如下为一条基本的sql语句：

select col1 from tabname where col2 > 10

QuerySpecificationContext节点下会产生用于扫描数据源的FromClauseContext、过滤条件对应的BooleanDefaultContext、以及投影时所需的NamedExpressionSeqContext节点。

1）FromClauseContext继续访问其子节点，当访问到TableINameContext节点时，访问到tableName的tocken时根据表名生成UnresolvedRelation：

override def visitTableName(ctx: TableNameContext): LogicalPlan = withOrigin(ctx) {
  val tableId = visitMultipartIdentifier(ctx.multipartIdentifier)
  val table = mayApplyAliasPlan(ctx.tableAlias, UnresolvedRelation(tableId))
  table.optionalMap(ctx.sample)(withSample)
}

2）BooleanDefaultContext的子节点中分为三个分支：代表Reference的ValueExpressionDefaultContext、代表数值的ValueExpressionDefaultContext、以及代表运算符的ComparisonContext；

例如遍历代表数据值ValueExpressionDefaultContext及其子节点，直到访问到IntegerLiteralContext:

override def visitIntegerLiteral(ctx: IntegerLiteralContext): Literal = withOrigin(ctx) {
  BigDecimal(ctx.getText) match {
    case v if v.isValidInt =>
      Literal(v.intValue)
    case v if v.isValidLong =>
      Literal(v.longValue)
    case v => Literal(v.underlying())
  }
}

而Literal的定义如下，是一个叶子类型的Expression节点：

case class Literal (value: Any, dataType: DataType) extends LeafExpression

3）NamedExpressionSeqContext是投影节点，迭代遍历直到RegularQuerySpecificationContext节点，然后通过访问withSelectQuerySpecification方法创建出投影所需的Project Logical Plan：

override def visitRegularQuerySpecification(
    ctx: RegularQuerySpecificationContext): LogicalPlan = withOrigin(ctx) {
  val from = OneRowRelation().optional(ctx.fromClause) {
    visitFromClause(ctx.fromClause)
  }
  withSelectQuerySpecification(
    ctx,
    ctx.selectClause,
    ctx.lateralView,
    ctx.whereClause,
    ctx.aggregationClause,
    ctx.havingClause,
    ctx.windowClause,
    from
  )
}
...
def createProject() = if (namedExpressions.nonEmpty) {
  Project(namedExpressions, withFilter)
} else {
  withFilter
}

总结一下以上处理过程中所涉及的类之间的关系，如下图所示：

生成Resolved LogicalPlan

Spark Analyser

在SparkSession的sql方法中，对sql语句进行过Parser解析并生成Unresolved LogicalPlan之后则通过执行Dataset.ofRows(self, plan, tracker) 继续进行catalog绑定，数据源绑定的过程如下：

def ofRows(sparkSession: SparkSession, logicalPlan: LogicalPlan, tracker: QueryPlanningTracker)
  : DataFrame = sparkSession.withActive {
  val qe = new QueryExecution(sparkSession, logicalPlan, tracker)
  qe.assertAnalyzed()
  new Dataset[Row](qe, RowEncoder(qe.analyzed.schema))
}
...
def assertAnalyzed(): Unit = analyzed

由如下实现逻辑可见， analyzed变量是通过懒加载方式初始化的，通过该变量的初始方法可见Spark的catalog实现逻辑主要通过Analyser类来实现的：

lazy val analyzed: LogicalPlan = executePhase(QueryPlanningTracker.ANALYSIS) {
  // We can't clone `logical` here, which will reset the `_analyzed` flag.
  sparkSession.sessionState.analyzer.executeAndCheck(logical, tracker)
}

其中，executeAndCheck方法的执行是通过Analyzer的父类RuleExecutor的execute方法来实现的：

def execute(plan: TreeType): TreeType = {
...
  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 startTime = System.nanoTime()
          val result = rule(plan)
          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)
          }
          ...
          result
      }
      iteration += 1
      if (iteration > batch.strategy.maxIterations) {
        // Only log if this is a rule that is supposed to run more than once.
        if (iteration != 2) {
          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) {
            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
      }

      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
}

如上代码的主要处理过程如下：

1、遍历的Analyzer类中的batches列表：

通过batches方法获取所有的catalog绑定相关的规则，在Analyzer中包括Substitution、Hints、Resolution、UDF、Subquery等几个规则组；

以较为常见的"Resolution"规则组为例，其具有非常多的规则用于解析函数、Namespace、数据表、视图、列等信息，当然用户也可以子定义相关规则：

Batch("Resolution", fixedPoint,
  ResolveTableValuedFunctions ::
  ResolveNamespace(catalogManager) ::
  new ResolveCatalogs(catalogManager) ::
  ResolveInsertInto ::
  ResolveRelations ::
  ResolveTables ::
  ResolveReferences ::
  ResolveCreateNamedStruct ::
  ResolveDeserializer ::
  ResolveNewInstance ::
  ResolveUpCast ::
  ResolveGroupingAnalytics ::
  ResolvePivot ::
  ResolveOrdinalInOrderByAndGroupBy ::
  ResolveAggAliasInGroupBy ::
  ResolveMissingReferences ::
  ...

其中，Batch类的定义如下，包括Batch名称、循环执行策略、具体的规则组集合，循环执行策略Strategy又分为Once和FixedPoint两种，即仅执行一次和固定次数：

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

2、将每个Batch中所有的规则Rule对象实施于该Unsolved LogicalPlan，并且该Batch中规则可能要执行多轮，直到执行的批数等于batch.strategy.maxIterations或者logicalplan与上个批次的结果比没有变化，则退出执行；

其中在Spark 中的定义如下，在spark3.0中默认可最大循环100次：

protected def fixedPoint =
  FixedPoint(
    conf.analyzerMaxIterations,
    errorOnExceed = true,
    maxIterationsSetting = SQLConf.ANALYZER_MAX_ITERATIONS.key)
...
val ANALYZER_MAX_ITERATIONS = buildConf("spark.sql.analyzer.maxIterations")
    .internal()
    .doc("The max number of iterations the analyzer runs.")
    .version("3.0.0")
    .intConf
    .createWithDefault(100)

接下来以将ResolveRelations（解析数据表或者视图）规则应用于Unresolved LogicalPlan的解析过程为例,支持解析UnresolvedRelation、UnresolvedTable、UnresolvedTableOrView等多种未解析的数据源：

def apply(plan: LogicalPlan): LogicalPlan = ResolveTempViews(plan).resolveOperatorsUp {
 ...
  case u: UnresolvedRelation =>
    lookupRelation(u.multipartIdentifier).map(resolveViews).getOrElse(u)

  case u @ UnresolvedTable(identifier) =>
    lookupTableOrView(identifier).map {
      case v: ResolvedView =>
        u.failAnalysis(s"${v.identifier.quoted} is a view not table.")
      case table => table
    }.getOrElse(u)

  case u @ UnresolvedTableOrView(identifier) =>
    lookupTableOrView(identifier).getOrElse(u)
}

当解析对象为UnresolvedRelation实例时，调用lookupRelation方法来对其进行解析，通过SessionCatalog或者扩展的CatalogPlugin来获取数据源的元数据，并生成Resolved LogicalPlan：

private def lookupRelation(identifier: Seq[String]): Option[LogicalPlan] = {
  expandRelationName(identifier) match {
    case SessionCatalogAndIdentifier(catalog, ident) =>
      lazy val loaded = CatalogV2Util.loadTable(catalog, ident).map {
        case v1Table: V1Table =>
          v1SessionCatalog.getRelation(v1Table.v1Table)
        case table =>
          SubqueryAlias(
            catalog.name +: ident.asMultipartIdentifier,
            DataSourceV2Relation.create(table, Some(catalog), Some(ident)))
      }
    ...

最常见的是SessionCatalog，作为SparkSession级别catalog接口对象，其定义如下，包括ExternalCatalog、GlobalTempViewManager、FunctionRegistry、SQLConf、Hadoop的Configuration、Parser、FunctionResourceLoader对象；其中，ExternalCatalog有两个主要的实现类：HiveExternalCatalog和InMemoryCatalog，而HiveExternalCatalog则主要应用于企业级的业务场景中:

class SessionCatalog(
    externalCatalogBuilder: () => ExternalCatalog,
    globalTempViewManagerBuilder: () => GlobalTempViewManager,
    functionRegistry: FunctionRegistry,
    conf: SQLConf,
    hadoopConf: Configuration,
    parser: ParserInterface,
    functionResourceLoader: FunctionResourceLoader) 

如果采用默认的SessionCatalog，当需要获取数据表时则通过ExternalCatalog实例调用其对应的接口来实现：

override def loadTable(ident: Identifier): Table = {
  val catalogTable = try {
    catalog.getTableMetadata(ident.asTableIdentifier)
  } catch {
    case _: NoSuchTableException =>
      throw new NoSuchTableException(ident)
  }

  V1Table(catalogTable)
}
...
def getTableMetadata(name: TableIdentifier): CatalogTable = {
    val db = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
    val table = formatTableName(name.table)
    requireDbExists(db)
    requireTableExists(TableIdentifier(table, Some(db)))
    externalCatalog.getTable(db, table)
 }

接下来如果采用ExternalCatalog接口的实现类HiveExternalCatalog的情况下，则通过HiveClientImpl类从Hive的metadata中类获取用户表的元数据相关信息：

private def getRawTableOption(dbName: String, tableName: String): Option[HiveTable] = {
  Option(client.getTable(dbName, tableName, false /* do not throw exception */))
}

另外，如需扩展的catalog范围可通过实现CatalogPlugin接口、并且配置“spark.sql.catalog.spark_catalog”参数来实现，例如在iceberg数据湖的实现中通过自定义其catalog来实现其个性化的逻辑：

spark.sql.catalog.spark_catalog = org.apache.iceberg.spark.SparkSessionCatalog

3、返回解析后的Resolved LogicalPlan。

以上处理逻辑中所涉及的主要的类之间的关系如下所示：

接下来仍然以前面的SQL语句（select col1 from tabname where col2 > 10）为例，简要阐述如何将一个Unresolved LogicalPlan解析成为Analyzed LogicalPlan：

1、根据Analyzer的解析规则，UnResolvedRelation节点可以应用到ResolveRelations规则，通过CatalogManger获取数据源中表的信息，得到Relation的相关列的信息并加上标号，同时创建一个针对数据表的SubqueryAlias节点；

2、针对过滤条件col2>10的过滤条件，针对列UnresolvedAttribute 可以适用到ResolveReference规则，根据第1步中得到的列信息可以进行解析；数字10可以应用到ImplicitTypeCasts 规则对该数字匹配最合适的数据类型；

3、针对Project 节点，接下来在进行下一轮解析，再次匹配到ResolveReference规则对投影列进行解析，从而将整棵树解析为Resolved LogicalPlan。

生成Optimized LogicalPlan

得到Resolved LogicalPlan之后，为了使SQL语句的执行性能更优，则需要根据一些规则进一步优化逻辑计划树，生成Optimized LogicalPlan。

本文采用的是Spark 3.0的源码，生成Optimized LogicalPlan是通过懒加载的方式被调用的，并且Optimizer类与Analyzer类一样继承了 RuleExecutor类，所有基于规则(RBO)的优化实际都是通过RuleExecutor类来执行，同样也是将所有规则构建为多个批次，并且将所有批次中规则应用于Analyzed LogicalPlan，直到树不再改变或者执行优化的循环次数超过最大限制（spark.sql.optimizer.maxIterations，默认100）：

lazy val optimizedPlan: LogicalPlan = executePhase(QueryPlanningTracker.OPTIMIZATION) {
  // clone the plan to avoid sharing the plan instance between different stages like analyzing,
  // optimizing and planning.
  val plan = sparkSession.sessionState.optimizer.executeAndTrack(withCachedData.clone(), tracker)
  // We do not want optimized plans to be re-analyzed as literals that have been constant folded
  // and such can cause issues during analysis. While `clone` should maintain the `analyzed` state
  // of the LogicalPlan, we set the plan as analyzed here as well out of paranoia.
  plan.setAnalyzed()
  plan
}
...
def executeAndTrack(plan: TreeType, tracker: QueryPlanningTracker): TreeType = {
    QueryPlanningTracker.withTracker(tracker) {
      execute(plan)
    }
}

逻辑计划优化规则仍然又多个Batch组成，每个Batch中包含多个具体的Rule并且可以执行一次或者固定次数。其中比较常用的优化规则有：谓词下推、常量累加、列剪枝等几种。

谓词下推将尽可能使得谓词计算靠近数据源，根据不同的场景有LimitPushDown、PushProjectionThroughUnion、PushDownPredicates等多种实现，  PushDownPredicates又包含PushPredicateThroughNonJoin和PushPredicateThroughJoin；

其中，PushPredicateThroughJoin可实现将谓词计算下推至join算子的下面，从而可以提升数据表之间的join计算过程中所带来的网络、内存以及IO等性能开销：

val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
  // push the where condition down into join filter
  case f @ Filter(filterCondition, Join(left, right, joinType, joinCondition, hint)) =>
    val (leftFilterConditions, rightFilterConditions, commonFilterCondition) =
      split(splitConjunctivePredicates(filterCondition), left, right)
    joinType match {
      case _: InnerLike =>
        // push down the single side `where` condition into respective sides
        val newLeft = leftFilterConditions.
          reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
        val newRight = rightFilterConditions.
          reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
        val (newJoinConditions, others) =
          commonFilterCondition.partition(canEvaluateWithinJoin)
        val newJoinCond = (newJoinConditions ++ joinCondition).reduceLeftOption(And)

        val join = Join(newLeft, newRight, joinType, newJoinCond, hint)
        if (others.nonEmpty) {
          Filter(others.reduceLeft(And), join)
        } else {
          join
        }
      case RightOuter =>
        // push down the right side only `where` condition

常量折叠是通过ConstantFolding规则来实现的，如果表达式中的算子是可以折叠的则在该阶段直接生成计算结果，以避免在实际的sql执行过程中产生逐行计算，从而可以降低CPU的计算开销：

object ConstantFolding extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    case q: LogicalPlan => q transformExpressionsDown {
      // Skip redundant folding of literals. This rule is technically not necessary. Placing this
      // here avoids running the next rule for Literal values, which would create a new Literal
      // object and running eval unnecessarily.
      case l: Literal => l

      // Fold expressions that are foldable.
      case e if e.foldable => Literal.create(e.eval(EmptyRow), e.dataType)
    }
  }
}

列剪枝规则通过ColumnPruning规则来实现，去掉不需要处理的列，可避免从数据源读取较多的数据列、将不需要的列加载至内存中计算计算计算中、以及返回不需要数据（想象一下大宽表的情况），从而获得较大的性能收益：

object ColumnPruning extends Rule[LogicalPlan] 
def apply(plan: LogicalPlan): LogicalPlan
prunedChild(c: LogicalPlan, allReferences: AttributeSet):LogicalPlan=
    if (!c.outputSet.subsetOf(allReferences)) {
    

Optimizer所涉及的主要类的关联关系如下图所示：

当所有优化规则完成对于Aanalyzed LogicalPlan的应用则可生成Optimized LogicalPlan。

本文重点讲解了Spark SQL解析为AST抽象语法树、生成Unresolved LogicalPlan、生成Resolved LogicalPlan以及Optimized LogicalPlan的过程，为接下来进一步生成物理计划Spark Plan做好了准备。

作者简介

焦媛，负责民生银行Hadoop大数据平台的生产运维工作，以及HDFS和Spark周边开源产品的技术支持，并致力于Spark云原生技术的支持与推广。