Spark SQL 工作流程源码解析（二）parsing 阶段（基于 Spark 3.3.0）_Shockang_spark sql 源码

网络 02-07 2040

前言

本文隶属于专栏《1000个问题搞定大数据技术体系》，该专栏为笔者原创，引用请注明来源，不足和错误之处请在评论区帮忙指出，谢谢！

本专栏目录结构和参考文献请见1000个问题搞定大数据技术体系

Spark SQL 工作流程源码解析（一）总览（基于 Spark 3.3.0）

Spark SQL 工作流程源码解析（二）parsing 阶段（基于 Spark 3.3.0）

Spark SQL 工作流程源码解析（三）analysis 阶段（基于 Spark 3.3.0）

Spark SQL 工作流程源码解析（四）optimization 阶段（基于 Spark 3.3.0）

Spark SQL 工作流程源码解析（五）planning 阶段（基于 Spark 3.3.0）

思维导图

正文 Hive SQL ? Spark SQL ? ANSI SQL

Spark SQL 是从 Shark 发展来的，而 Shark 又被称为“Hive on Spark”，相比 ANSI SQL（国际标准 SQL），可能 Spark SQL 更偏向于 Hive SQL 一点（吐槽一句，Hive 3.x 版本目前兼容性是比较差的）。

你可以通过我的这篇博客了解详情——Spark SQL是怎么发展起来的？

Spark 3.x 新增了一个参数：spark.sql.ansi.enabled，默认情况下是 false，表示采取和 Hive 一样的方式来对待 SQL，当它被设置成 true 的时候，就会采取 ANSI SQL（2011）的方式。比如一个整型或者小数字段发生溢出情况的话，Spark 会抛出一个运行时异常。并且，Spark 会在 SQL 编译器里面禁止使用 ANSI SQL 的保留字。

而且，Spark SQL 是逐渐朝着 ANSI SQL 靠拢的，这也是 SQL 引擎的发展趋势。

此时，就面临一个问题，SQL 引擎要支持的语法实际上在不停的变化，那么是不是就意味着对应的编译器代码也得不停的改变呢？这样改动起来会不会太麻烦了点啊？

要明白这个问题，先得了解一下 SQL 是如何编译的？

DSL 编译

Spark SQL 本质上只是一个 DSL（领域专用语言），它代表的语法规则实际上只是 Spark 领域独有的。

而 DSL 的构建与通用编程语言的构建类似，主要的过程仍然是指定语法和语义，然后实现编译器或解释器。

通常情况下，一个系统中 DSL 模块的实现需要涉及两方面的工作。

设计语法和语义，定义 DSL 中具体的元素。实现词法分析器（Lexer）和语法器（Parser），DSL的解析，最终为底层逻辑来执行。

经过几十年的研究，编译理论是比较成熟的，借助于各种工具，开发人员不需要从头开始构建烦琐的词法分析和语法分析模块｡

迄今为止，业界提供了各种各样生成器，可以直接应用在系统中。

基于生成器，实现一个编译器前端就像“填表”那么简单，只要提供特定的文法即可。

而 ANTLR 就是这样的一款工具，基本上大数据领域知名的 SQL 引擎都使用了它，比如：Hive，Spark，Flink 等。

ANTLR

Spark 从 2.0版本开始就使用 ANTLR ( Another Tool for Language Recognition) 进行词法和语法解析，它为Java、C++ 和 C# 等语言提供了一个通过语法描述来自动构造语言的识别器（Recognizer）、编译器（Parser）和解释器（Translator）的框架。

如果你下载 Apache Spark 源码后出现了编译问题，实际上就和 ANTLR 这个工具有关，你可以查看我的这篇博客了解详情——编译 Apache Spark 源码报错？那是因为你漏掉了关键操作

当面临开发新的语法支持时，首先需要改动的是 ANTLR 4 文件（在 SqlBase.g4 中添加文法），重新生成词法解析器（SqlBaseLexer）、语法解析器（SqlBaseParser）和访问者接口（SqlBaseVisitor）与访问者类（SqlBaseBaseVisitor）, 然后在 AstBuilder 等类中添加相应的访问逻辑，最后添加执行逻辑。

这里也就回答了上面的问题，使用了 ANTLR 4，我们通常只需要改动很少的地方，就能非常方便地添加各种语法。

Unresolved Logical Plan

Spark 使用 ANTLR 将 SQL/DataFrame/Dataset 转化成 AST（抽象语法树），这个 AST 由于未绑定具体的类型，所以通常称为 Unresolved Logical Plan（这个称呼来源于 Spark SQL 的原始论文，源码中一般称为 Parsed Logical Plan）。

下面我们来看看具体的源码实现

parsing 阶段的入口 ParseDriver.scala 这个文件里面

AbstractSqlParser.parsePlan /** 为给定的 SQL 字符串创建一个 LogicalPlan */ 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 QueryParsingErrors.sqlStatementUnsupportedError(sqlText, position) } }

看起来比较简单的一段代码实际上包含下面 4 个流程阶段：

parse生成解析树生成 AST结果处理

具体的 SQL 解析在 parse 方法里面

1、AbstractSqlParser.parse protected def parse[T](command: String)(toResult: SqlBaseParser => T): T = { logDebug(s"Parsing command: $command") // SqlBase.g4 生成的词法解析器 // 这里会将 SQL 命令转化成不区分大小写的字符流传递给词法分析器 val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command))) // 清空用来识别错误的监听器列表 lexer.removeErrorListeners() // 添加自定义的编译错误监听器 lexer.addErrorListener(ParseErrorListener) // token 流指定来源 val tokenStream = new CommonTokenStream(lexer) // SqlBase.g4 生成的语法解析器 val parser = new SqlBaseParser(tokenStream) // 语法解析器添加后置处理器，专门用来验证并清理解析树 parser.addParseListener(PostProcessor) // 添加后置处理器用来检查未闭合括号的注释的 parser.addParseListener(UnclosedCommentProcessor(command, tokenStream)) // 同上，先清空用来识别错误的监听器列表 parser.removeErrorListeners() // 同上，添加自定义的编译错误监听器 parser.addErrorListener(ParseErrorListener) // 如果为false，则根据SQL标准，INTERSECT的优先级高于其他集合操作 //（ UNION，EXCEPT 和 MINUS ）。 parser.legacy_setops_precedence_enabled = conf.setOpsPrecedenceEnforced // 如果为false，则带有指数的文本将转换为double类型而不是decimal类型。 parser.legacy_exponent_literal_as_decimal_enabled = conf.exponentLiteralAsDecimalEnabled // 如果为true，则关键字的行为遵循ANSI SQL标准。 parser.SQL_standard_keyword_behavior = conf.enforceReservedKeywords try { try { // 先使用 ANTLR 较快的 SLL 模式进行解析，成功返回结果 parser.getInterpreter.setPredictionMode(PredictionMode.SLL) toResult(parser) } catch { case e: ParseCancellationException => // 如果解析失败，复位 tokenStream.seek(0) // 把输入流的索引改成 0，倒带输入流 parser.reset() // 重试，再使用 LL 模式进行解析，成功返回结果 parser.getInterpreter.setPredictionMode(PredictionMode.LL) toResult(parser) } } catch { // 编译异常处理 case e: ParseException if e.command.isDefined => throw e case e: ParseException => throw e.withCommand(command) case e: AnalysisException => val position = Origin(e.line, e.startPosition) throw new ParseException(Option(command), e.message, position, position, e.errorClass, e.messageParameters) } }

具体做的事情，上面的注释已经写的很详细了，这里再给出流程图：

2、生成解析树 SqlBaseParser.singleStatement

不同版本 SqlBase.g4 文件生成的语法解析器 SqlBaseParser 是不一样的，但是逻辑都是类似的。下面的源码直接来自 Apache Spark 的 master 分支（目前是 3.3.0-SNAPSHOT）

public final SingleStatementContext singleStatement() throws RecognitionException { // 定义一个单语句场景的上下文 SingleStatementContext _localctx = new SingleStatementContext(_ctx, getState()); // 告诉监听器我要进入规则的具体执行了 enterRule(_localctx, 0, RULE_singleStatement); int _la; try { // 设置上下文节点的外部备选编号 enterOuterAlt(_localctx, 1); { // 指示识别器已更改与传入的ATN状态一致的内部状态。 // 这样，随着解析器的运行，我们总是知道我们在ATN中的位置。 // 规则上下文对象形成一个堆栈，让我们可以看到调用规则的堆栈。 // 结合这一点，我们有完整的ATN配置信息。 // 下同，不再注释。 setState(294); // 核心逻辑 statement(); setState(298); // 这里为错误处理程序提供了在输入流中的语法或语义错误导致识别异常之前处理这些错误的机会。 _errHandler.sync(this); // 获取从当前位置偏移1处的符号值 _la = _input.LA(1); while (_la==T__0) { { { setState(295); // 将当前输入符号与 T__0 匹配 match(T__0); } } setState(300); _errHandler.sync(this); _la = _input.LA(1); } setState(301); // 匹配结束字符 match(EOF); } } catch (RecognitionException re) { _localctx.exception = re; _errHandler.reportError(this, re); _errHandler.recover(this, re); } finally { // 告诉监听器我要结束规则的执行了 exitRule(); } return _localctx; }

扩充转移网络(ATN)，是 Bill Woods 在 1970 年提出的一种分析器。在那之后，ATN 在自然语言分析领域中作为一种形式化方法，被广为使用。

SqlBaseParser.statement public final StatementContext statement() throws RecognitionException { // 创建一个语句上下文 StatementContext _localctx = new StatementContext(_ctx, getState()); // 告诉监听器我进来啦 enterRule(_localctx, 14, RULE_statement); int _la; try { int _alt; setState(1110); _errHandler.sync(this); // 获取识别器用于预测的ATN解释器，判断是什么类型的语句 switch ( getInterpreter().adaptivePredict(_input,117,_ctx) ) { // 查询语句 case 1: _localctx = new StatementDefaultContext(_localctx); enterOuterAlt(_localctx, 1); { setState(321); // 查询的核心逻辑 query(); } break; ...

查询的核心逻辑和上面的代码都很类似，考虑到篇幅问题，这里就不一行行代码讲解了。

解析树到底长啥样子？

上面的解析树就是我们在第一讲——Spark SQL 工作流程源码解析（一）总览（基于 Spark 3.3.0）中给出的例子最终形成的解析树。

TerminalNodeImpl 来自 ANTLR 4 的 jar 包，代表的就是一个个末端节点。

3、生成 AST

在看这部分的源码前首先要明白什么是访问者模式，为什么要用访问者模式。

访问者模式

访问者模式，就是在一个数据结构相对稳定的系统上面定义各种操作，它解耦了数据结构和作用于结构上面的操作。

如果一个系统有比较稳定的数据结构，又有易于变化的算法的话，使用访问者模式是比较合适的。

访问者模式的优点是增加新的操作很容易，因为增加新的操作就意味着增加一个新的访问者，访问者模式会把有关的行为集中到一个访问者对象中。

访问者模式的缺点就是增加新的数据结构会变得困难。

那么，访问者模式该怎样实现呢？

我们结合 Apache Spark 的源码来学习访问者的实现。

类图

上面的类图分为 3 个模块，antlr 代表来自 ANTLR v4，spark 代表来自 Apache Spark，gen 代表 SqlBase.g4 生成的。

3 种颜色代表访问者模式的 3 种不同角色：

红色 —— 数据结构绿色 —— 访问者橙色 —— 导游（Traverser）

上图中只展示了核心代码的类图，一些不太重要的没有画出来，建议跟着 Apache Spark 的源码来理解上面的类图。

上图中数据结构只画了 SingleStatementContext 这个根节点，实际上本类图要结合上面的解析图来一起理解，上面的解析树都是属于数据结构的部分。

我们简单讲解一下上面的类图：

数据结构

ParseTree 是解析树接口，解析树中类似 SingleStatementContext 这样节点都是它的实现类。

ParseTree 中定义了一个 accept 接口方法，接受一个 ParseTreeVisitor 对象，这是典型的访问者模式的应用。

<T> T accept(ParseTreeVisitor<? extends T> visitor);

SingleStatementContext 中有 statement、enterRule、exitRule、accept 等方法，这个前面的源码解析中也提到过，statement 返回节点信息的上下文对象，enterRule、exitRule 都是用于和监听器联动，在进入规则和退出规则时调用监听器的相应方法，accept 实现了 ParseTree 的接口方法。

访问者

ParseTreeVisitor 代表的是解析树ParseTree的访问者接口对象，其中定义了访问 ParseTree 中各种类型节点的接口方法。

ANTLR v4 自己提供了一个抽象类 AbstractParseTreeVisitor 实现了接口ParseTreeVisitor ，visit 方法的默认实现会调用解析树的 accept 方法，并将当前访问者对象的引用传递进去，如下所示：

@Override public T visit(ParseTree tree) { return tree.accept(this); }

实际上这里体现了双重分派（double dispatch）的思想，ParseTree 和ParseTreeVisitor 都有众多的实现类，上面的代码使得每个相应的解析树节点对象（一重）在调用 accept 方法的时候，会根据访问者的具体类型选择具体的访问方法（二重）。

这里的访问方法指的是访问不同类型的节点有相应的处理，每种处理方式都会对应一个方法。

有人可能就问了，这不就太麻烦了吗？

Spark SQL 支持的语法是在不断升级的，每支持一种语法我都添加一个对应的访问方法，我哪里改的过来啊~

所以，这里就体现出 ANTLR 的好处了，我们只需要在 SqlBase.g4 文件里面统一变动，这些访问方法机器会帮我们搞定！

SqlBase.g4文件生成的访问者接口是SqlBaseVisitor，并且同时生成了一个默认的实现类SqlBaseBaseVisitor。

在具体的访问方法里面，默认会调用其子节点具体的访问方法，如下所示：

@Override public T visitSingleStatement(SqlBaseParser.SingleStatementContext ctx) { return visitChildren(ctx); }

Spark 在 AstBuilder 中重写了SqlBaseBaseVisitor的一些访问方法，这个类主要适用于 Catalyst 内部调用，而外部调用则由AstBuilder的子类SparkSqlAstBuilder负责。

细心的同学已经发现了，我们在第一讲中提到过 CatalystSqlParser 用于 Catalyst 内部调用，而 SparkSqlParser 用于外部调用，和上面的描述多像啊~

实际上SparkSqlParser 和 CatalystSqlParser 在访问者模式中充当了导游的作用，它们是引领着访问者进行访问操作的，具体的怎么回事呢？且听我慢慢道来~

导游

Spark 中扮演导游身份的是从接口ParserInterface开始的，我们在第一讲中也提到过这个接口，这里不在赘述了。

Spark 将一些通用的方法属性放到了抽象实现类 AbstractSqlParser 中，这个类想必大家比较熟悉了，parsing 阶段的入口就在这个类中，这也很好理解，不论是外部调用还是内部调用，我都走一个统一的入口嘛，而SparkSqlParser 和 CatalystSqlParser就是AbstractSqlParser 的 2 个子类。

其中，值得注意的是：

AbstractSqlParser 中定义了一个抽象方法：

protected def astBuilder: AstBuilder

SparkSqlParser 中实现了这个抽象方法（在 Scala 中不带括号的方法等同于属性）

val astBuilder = new SparkSqlAstBuilder()

CatalystSqlParser 中这样来实现：

val astBuilder = new AstBuilder

看到这里，是不是和前面的内容联系起来了呢~

说到底，不论是SparkSqlParser 还是 CatalystSqlParser，都是对访问者SqlBaseVisitor（SparkSqlParser 中是SparkSqlAstBuilder，CatalystSqlParser中是AstBuilder）的一个封装而已，底层干活的都是它，这就和上面说的内外部调用不矛盾啦。

也可以看出来，Spark 的外部调用（SQL/DataFrame/Dataset）只是在Catalyst 内部调用基础上面的一个拓展，因为SparkSqlParser是AstBuilder的子类啊~

看到这里，你再看看前面的类图，是不是有种豁然开朗的感觉呢~

走读源码

我先来带领大家走读一下源码，和前面讲的内容串联起来。

解析树的根节点是SingleStatementContext，所以我们先调用访问者的visitSingleStatement方法，这个方法在AstBuilder.scala 里面。

AstBuilder.visitSingleStatement override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan = withOrigin(ctx) { visit(ctx.statement).asInstanceOf[LogicalPlan] } /** * 注册上下文的来源。 * 在闭包中创建的任何 TreeNode 都将被指定为注册源。 * 此方法在完成闭包后恢复先前设置的来源。 */ def withOrigin[T](ctx: ParserRuleContext)(f: => T): T = { val current = CurrentOrigin.get CurrentOrigin.set(position(ctx.getStart)) try { f } finally { CurrentOrigin.set(current) } }

CurrentOrigin 为 TreeNode 提供了一个位置，以便询问其来源的上下文。例如，当前正在分析哪一行代码。

SingleStatementContext.statement public StatementContext statement() { // 就是查找 SingleStatementContext 的子节点中第一个类型是 StatementContext 的节点 return getRuleContext(StatementContext.class,0); } public <T extends ParserRuleContext> T getRuleContext(Class<? extends T> ctxType, int i) { return getChild(ctxType, i); } public <T extends ParseTree> T getChild(Class<? extends T> ctxType, int i) { if ( children==null || i < 0 || i >= children.size() ) { return null; } int j = -1; // 找到指定类型的子节点 for (ParseTree o : children) { if ( ctxType.isInstance(o) ) { j++; if ( j == i ) { return ctxType.cast(o); } } } return null; }

RuleContext.accept

访问 StatementContext 实际上调用的是 RuleContext 的默认实现方法：

@Override public <T> T accept(ParseTreeVisitor<? extends T> visitor) { return visitor.visitChildren(this); }

AstBuilder.visitChildren /** * 覆盖所有访问方法的默认行为。只有当上下文只有一个子项时，才会返回非空结果。 * 之所以这样做，是因为没有通用方法来组合上下文子项的结果。在所有其他情况下，返回null。 */ override def visitChildren(node: RuleNode): AnyRef = { if (node.getChildCount == 1) { node.getChild(0).accept(this) } else { null } } @Override public ParseTree getChild(int i) { return children!=null && i>=0 && i<children.size() ? children.get(i) : null; }

结合上面的解析树可以看到子节点就一个—— QueryContext

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

AstBuilder.visitQuery override def visitQuery(ctx: QueryContext): LogicalPlan = withOrigin(ctx) { val query = plan(ctx.queryTerm).optionalMap(ctx.queryOrganization)(withQueryResultClauses) // Apply CTEs query.optionalMap(ctx.ctes)(withCTE) }

CTE（Common Table Expression，公用表表达式）是定义在SELECT、INSERT、UPDATE或DELETE语句中的临时命名的结果集，同时CTE也可以用在视图的定义中。公用表表达式提供的功能其实和视图差不多，但是它不像视图一样把SQL语句保存在我们的数据库里面。使用CTE可以把复杂的SQL语句按照逻辑分成简单独立的几个公用表表达式（CTE），这样的最大优势就是能够提高SQL语句的可读性和可维护性。总结就是，CTE主要可以用于树结构的递归和简化SQL语句，增加可读性和可维护性。

public QueryTermContext queryTerm() { return getRuleContext(QueryTermContext.class,0); }

这是在 QueryContext 的子节点中查找第一个是 QueryTermContext 类型的。

protected def plan(tree: ParserRuleContext): LogicalPlan = typedVisit(tree) protected def typedVisit[T](ctx: ParseTree): T = { ctx.accept(this).asInstanceOf[T] }

可以看到 AstBuilder 的 plan 方法就是继续调用 QueryTermContext 的 accept 方法，这个方法调用的还是 AstBuilder.visitChildren 方法。

/** * 如果存在传递的上下文，则将一个LogicalPlan映射到另一个LogicalPlan。 * 当上下文不存在时，返回原始计划。 */ def optionalMap[C](ctx: C)(f: (C, LogicalPlan) => LogicalPlan): LogicalPlan = { if (ctx != null) { f(ctx, plan) } else { plan } } /** * 添加 ORDER BY/SORT BY/CLUSTER BY/DISTRIBUTE BY/LIMIT/WINDOWS 子句到逻辑计划中。子句决定了查询结果的 `shape` (ordering/partitioning/rows) */ private def withQueryResultClauses( ctx: QueryOrganizationContext, query: LogicalPlan): LogicalPlan = withOrigin(ctx) { import ctx._ // 处理 ORDER BY, SORT BY, DISTRIBUTE BY 和 CLUSTER BY 子句。 val withOrder = if ( !order.isEmpty && sort.isEmpty && distributeBy.isEmpty && clusterBy.isEmpty) { // ORDER BY ... Sort(order.asScala.map(visitSortItem).toSeq, global = true, query) } else if (order.isEmpty && !sort.isEmpty && distributeBy.isEmpty && clusterBy.isEmpty) { // SORT BY ... Sort(sort.asScala.map(visitSortItem).toSeq, global = false, query) } else if (order.isEmpty && sort.isEmpty && !distributeBy.isEmpty && clusterBy.isEmpty) { // DISTRIBUTE BY ... withRepartitionByExpression(ctx, expressionList(distributeBy), query) } else if (order.isEmpty && !sort.isEmpty && !distributeBy.isEmpty && clusterBy.isEmpty) { // SORT BY ... DISTRIBUTE BY ... Sort( sort.asScala.map(visitSortItem).toSeq, global = false, withRepartitionByExpression(ctx, expressionList(distributeBy), query)) } else if (order.isEmpty && sort.isEmpty && distributeBy.isEmpty && !clusterBy.isEmpty) { // CLUSTER BY ... val expressions = expressionList(clusterBy) Sort( expressions.map(SortOrder(_, Ascending)), global = false, withRepartitionByExpression(ctx, expressions, query)) } else if (order.isEmpty && sort.isEmpty && distributeBy.isEmpty && clusterBy.isEmpty) { // [EMPTY] query } else { throw QueryParsingErrors.combinationQueryResultClausesUnsupportedError(ctx) } // WINDOWS val withWindow = withOrder.optionalMap(windowClause)(withWindowClause) // LIMIT // - LIMIT ALL is the same as omitting the LIMIT clause withWindow.optional(limit) { Limit(typedVisit(limit), withWindow) } } private def withCTE(ctx: CtesContext, plan: LogicalPlan): LogicalPlan = { val ctes = ctx.namedQuery.asScala.map { nCtx => val namedQuery = visitNamedQuery(nCtx) (namedQuery.alias, namedQuery) } // 检查重复命名 val duplicates = ctes.groupBy(_._1).filter(_._2.size > 1).keys if (duplicates.nonEmpty) { throw QueryParsingErrors.duplicateCteDefinitionNamesError( duplicates.mkString("'", "', '", "'"), ctx) } UnresolvedWith(plan, ctes.toSeq) }

由此，可以得出结论：

AstBuilder.visitQuery 方法就是结合 QueryContext 的 3 个子节点 QueryTermContext，QueryOrganizationContext，CtesContext 的信息进行包括 ORDER BY/SORT BY/CLUSTER BY/DISTRIBUTE BY/LIMIT/WINDOWS 子句和 CTE 的处理，其中 QueryTermContext 是信息的主要承载方，默认是其子类 QueryTermDefaultContext 来实例化。

我们的解析树中并没有包含 CTE 的处理。

带大家走读了详细的源码，接下来，基于篇幅限制，这里只给出部分关键地方的源码：

一、主流程 1、AstBuilder.visitRegularQuerySpecification override def visitRegularQuerySpecification( ctx: RegularQuerySpecificationContext): LogicalPlan = withOrigin(ctx) { val from = OneRowRelation().optional(ctx.fromClause) { // 处理 FROM 子句 visitFromClause(ctx.fromClause) } withSelectQuerySpecification( ctx, ctx.selectClause, ctx.lateralView, ctx.whereClause, ctx.aggregationClause, ctx.havingClause, ctx.windowClause, from ) } 2、AstBuilder.withSelectQuerySpecification /** * 向逻辑计划添加常规（SELECT）查询规范。 * 查询规范是逻辑计划的核心，就是寻源（FROM子句）、投影（SELECT）、聚合（GROUP BY…HAVING…）过滤 * （WHERE）。 * 请注意，查询 hints 会被忽略（解析器和生成器都会忽略）。 */ private def withSelectQuerySpecification( ctx: ParserRuleContext, selectClause: SelectClauseContext, lateralView: java.util.List[LateralViewContext], whereClause: WhereClauseContext, aggregationClause: AggregationClauseContext, havingClause: HavingClauseContext, windowClause: WindowClauseContext, relation: LogicalPlan): LogicalPlan = withOrigin(ctx) { // 是否需要去重 val isDistinct = selectClause.setQuantifier() != null && selectClause.setQuantifier().DISTINCT() != null val plan = visitCommonSelectQueryClausePlan( relation, // 处理 SELECT 子句 visitNamedExpressionSeq(selectClause.namedExpressionSeq), lateralView, whereClause, aggregationClause, havingClause, windowClause, isDistinct) // Hint selectClause.hints.asScala.foldRight(plan)(withHints) } 3、AstBuilder.visitCommonSelectQueryClausePlan def visitCommonSelectQueryClausePlan( relation: LogicalPlan, expressions: Seq[Expression], lateralView: java.util.List[LateralViewContext], whereClause: WhereClauseContext, aggregationClause: AggregationClauseContext, havingClause: HavingClauseContext, windowClause: WindowClauseContext, isDistinct: Boolean): LogicalPlan = { // 结合 FROM 子句和侧视图一起处理（在我们的例子中侧视图为空） val withLateralView = lateralView.asScala.foldLeft(relation)(withGenerate) // 流水线式的继续处理 WHERE 子句（在我们的例子中 WHERE 子句为空） val withFilter = withLateralView.optionalMap(whereClause)(withWhereClause) val namedExpressions = expressions.map { case e: NamedExpression => e case e: Expression => UnresolvedAlias(e) } def createProject() = if (namedExpressions.nonEmpty) { // 将上面的处理结果封装到 Project 对象之中 Project(namedExpressions, withFilter) } else { withFilter } // 先处理 Project val withProject = if (aggregationClause == null && havingClause != null) { if (conf.getConf(SQLConf.LEGACY_HAVING_WITHOUT_GROUP_BY_AS_WHERE)) { // 如果这个参数设置了，就把没有 GROUP BY 的 HAVING 当成 WHERE val predicate = expression(havingClause.booleanExpression) match { case p: Predicate => p case e => Cast(e, BooleanType) } Filter(predicate, createProject()) } else { // 根据 SQL 标准，没有 GROUP BY 的 HAVING 就意味着全局的聚合。 withHavingClause(havingClause, Aggregate(Nil, namedExpressions, withFilter)) } } else if (aggregationClause != null) { val aggregate = withAggregationClause(aggregationClause, namedExpressions, withFilter) aggregate.optionalMap(havingClause)(withHavingClause) } else { // 当命中这个分支的时候，HAVING 必须为 null。 createProject() } // 接着处理 Distinct val withDistinct = if (isDistinct) { // 可以看到，如果有 Distinct 就是将 Project 结果封装进 Distinct 对象之中 Distinct(withProject) } else { // 没有 Distinct 继续使用 Project 对象 withProject } // 最后处理 Window val withWindow = withDistinct.optionalMap(windowClause)(withWindowClause) // 如果没有 Distinct 和 Window 最终返回的还是 Project 对象 withWindow }

可以看到，这段代码是个很明显的流水线加工过程，整体流程如下：

二、处理 SELECT 子句 AstBuilder.visitNamedExpressionSeq override def visitNamedExpressionSeq( ctx: NamedExpressionSeqContext): Seq[Expression] = { Option(ctx).toSeq .flatMap(_.namedExpression.asScala) .map(typedVisit[Expression]) }

可以看到，SELECT 子句的处理结果就是一个命名表达式（NamedExpression）的集合

AstBuilder.visitStar /** * 创建一个星形（即全部）表达式；这将选择（指定对象中的）所有元素。 * 同时支持非目标（全局）别名和目标别名。 */ override def visitStar(ctx: StarContext): Expression = withOrigin(ctx) { UnresolvedStar(Option(ctx.qualifiedName()).map(_.identifier.asScala.map(_.getText).toSeq)) }

其中 UnresolvedStar 的类定义如下:

case class UnresolvedStar(target: Option[Seq[String]]) extends Star with Unevaluable

UnresolvedStar表示给定关系运算符的所有输入属性，例如在“SELECT * FROM …”中。

也适用于扩展结构体。

例如：

SELECT record.* from (SELECT struct(a,b,c) as record ...

形参 target 是扩展目标的可选名称。如果省略，则生成所有目标的列。可以是表名或结构体名称。这是扩展路径的标识符列表。

AbstractParseTreeVisitor.visitTerminal /** * 访问终端节点，并返回用户定义的操作结果。 * 默认实现返回defaultResult的结果。 * 指定的: 接口 ParseTreeVisitor中的探访者 * 形参: node–要访问的终端节点。 * 返回值: 访问节点的结果。 */ @Override public T visitTerminal(TerminalNode node) { return defaultResult(); } /** * 获取访问者方法返回的默认值。 * 该值由 visitTerminal、visitErrorNode 的默认实现返回。 * visitChildren的默认实现将其聚合结果初始化为该值。 * 基本实现返回null。 * 返回值: 访问者方法返回的默认值。 */ protected T defaultResult() { return null; } 三、处理 FROM 子句 /** * 为给定的“FROM”子句创建逻辑计划。 * 请注意，我们在这里支持多个（逗号分隔）关系，这些关系通过无条件内部联接转换为单个计划。 */ override def visitFromClause(ctx: FromClauseContext): LogicalPlan = withOrigin(ctx) { val from = ctx.relation.asScala.foldLeft(null: LogicalPlan) { (left, relation) => // 继续调用子节点的访问方法 val right = plan(relation.relationPrimary) // join 处理 val join = right.optionalMap(left) { (left, right) => if (relation.LATERAL != null) { if (!relation.relationPrimary.isInstanceOf[AliasedQueryContext]) { throw QueryParsingErrors.invalidLateralJoinRelationError(relation.relationPrimary) } LateralJoin(left, LateralSubquery(right), Inner, None) } else { Join(left, right, Inner, None, JoinHint.NONE) } } withJoinRelations(join, relation) } // pivot （即行转列）处理 if (ctx.pivotClause() != null) { if (!ctx.lateralView.isEmpty) { throw QueryParsingErrors.lateralWithPivotInFromClauseNotAllowedError(ctx) } withPivot(ctx.pivotClause, from) } else { ctx.lateralView.asScala.foldLeft(from)(withGenerate) } } AstBuilder.visitRelation /** * 创建FROM子句中引用的单个关系。join 条件的一部分被嵌套时会使用这个方法，例如： * {{{ * select * from t1 join (t2 cross join t3) on col1 = col2 * }}} */ override def visitRelation(ctx: RelationContext): LogicalPlan = withOrigin(ctx) { // 由于我们的例子中没有涉及到 join，此处不深入讲解 withJoinRelations(plan(ctx.relationPrimary), ctx) } AstBuilder.visitTableName /** * 创建一个别名表引用。这通常用于FROM子句。 */ override def visitTableName(ctx: TableNameContext): LogicalPlan = withOrigin(ctx) { val tableId = visitMultipartIdentifier(ctx.multipartIdentifier) val relation = UnresolvedRelation(tableId) // 如果 FROM 子句中指定了别名，就会为逻辑计划创建子查询别名和列别名 val table = mayApplyAliasPlan( ctx.tableAlias, relation.optionalMap(ctx.temporalClause)(withTimeTravel)) // 将样本添加到逻辑计划中 table.optionalMap(ctx.sample)(withSample) }

生成 AST 小结

至此，我们完成了生成 AST 部分核心源码的走读，下图是核心流程图：

这部分内容由于双重分派机制的原因，看起来非常的杂乱，方法调用在不同类之间不停的跳动。

但是细细思考，就会发现，本质上，生成 AST 就是对上面生成的解析树中每个节点，调用它们各自的访问方法（即visitXX）。

为什么要调用访问方法呢？

我们要明白，解析树虽然已经生成了，但是 Spark 程序对它还是无感的，按照我们人类的思维就是：

我知道这是个树，但是这棵树中每个节点具体是干嘛的，对我接下来的处理有啥作用，我都不清楚。

经过我们的处理之后，最终形成的树结构的逻辑算子树，SQL 中所包含的各种处理逻辑（过滤、剪裁等）和数据信息都会被整合在逻辑算子树的不同节点中。

相信，很多同学都对此时形成的 AST 感兴趣，这里是我们的例子逻辑计划的打印结果：

'Project [*] +- 'UnresolvedRelation [t_user], [], false

我们很容易就在上面的源码中找到对应的处理逻辑

打印对应的源码说明ProjectProject(namedExpressions, withFilter)Project 封装，其中 namedExpressions 为 SELECT 子句的处理结果，withFilter 为 FROM 子句的处理结果，且 withFilter 构成了当前逻辑计划的子节点，所以会出现在第二层[*]UnresolvedStar表示给定关系运算符的所有输入属性UnresolvedRelationUnresolvedRelationFROM 子句的处理结果[t_user]UnresolvedRelation.name表名[]UnresolvedRelation.output逻辑计划输出falseUnresolvedRelation.resolved表示当前逻辑计划是否已经被解析

3、结果处理

这部分的源码很简单：

case plan: LogicalPlan => plan case _ => val position = Origin(None, None) throw QueryParsingErrors.sqlStatementUnsupportedError(sqlText, position)

就是判断生成的逻辑计划是不是合法的，合法的正常返回，非法的就抛出包含 SQL字符串、出错行号和起始位置的 ParseException 异常信息。

总结

本讲是对 Spark SQL 工作流程中 parsing 阶段的源码解析。

我们从 ANTLR 这个工具入手，讲了这个工具是什么以及为什么会使用这个工具。

接着，我们将 parsing 划分成 4 个流程阶段：

其中，我们结合第一讲中给出的例子，画出了解析树。

然后我们提到了访问者模式，并结合 Apache Spark 的源码类图来理解访问者模式。

最后，我们走读了整个 parsing 阶段的核心源码，并结合最终逻辑计划的打印输出来深层次的理解源码。

至此，parsing 阶段的基本流程相信大家已经一览无余了。