spark sql 从antlr的ast到Unresolved Logical Plan

spark sql 从antlr的ast到Unresolved Logical Plan

前提

• 了解spark sql流程
• 了解antlr,能看懂antlr的语法文件(*.g4),了解antlr访问者模式,安装idea antlr插件

准备

生成antlr语法树

• 在idea中打开SqlBase.g4,这个是antlr规定的spark sql的sql语法.如果我们想看看某个关键字在spark sql 中怎么用,如果熟悉antlr的话,也可以到这个文件中查看.
• 编写sql

/* 在插件中所有字母必需大写
fragment LETTER: [A-Z];*/
SELECT NAME,AGE FROM USER

spark 中编写相应的代码,方便调试

import org.apache.spark.sql.SparkSessionobject TestSql {def main(args: Array[String]): Unit = {val spark: SparkSession = SparkSession.builder().config("spark.sql.shuffle.partitions", 5).master("local[*]").appName("us").getOrCreate()import spark._sql("SELECT NAME,AGE FROM USER").show()}}

  /** Creates LogicalPlan for a given SQL string. *///追踪代码,在这里把 sql转成了LogicalPlanoverride def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>astBuilder.visitSingleStatement(parser.singleStatement()) match {case plan: LogicalPlan => plancase _ =>val position = Origin(None, None)throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position)}}

• 打开类

class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging

这个类extends 了antlr的SqlBaseBaseVisitor,SqlBaseBaseVisitor对所有的方法都有一个默认的实现,即访问子节点.所以AstBuilder并不需要重新实现全部的方法,只用实现关心的方法.

public class SqlBaseBaseVisitor<T> extends AbstractParseTreeVisitor<T> implements SqlBaseVisitor<T> {/*** {@inheritDoc}** <p>The default implementation returns the result of calling* {@link #visitChildren} on {@code ctx}.</p>*/@Override public T visitSingleStatement(SqlBaseParser.SingleStatementContext ctx) { return visitChildren(ctx); }

• 开始分析
  结合上图生成的ast,第一个访问的节点是 singleStatement,ast中的

  /*第1步 singleStatement  语法树的起点*/override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan = withOrigin(ctx) {//visit 直接去访问子节点,由图可知下一个是statement:statementDefault(看冒号扣面的,给每个分支定义一个名称)visit(ctx.statement).asInstanceOf[LogicalPlan]}

visitStatementDefault没有实现,直接访问下一个节点

  /*** Create a top-level plan with Common Table Expressions.* query* : ctes? queryNoWith* 第2步* ;*/override def visitQuery(ctx: QueryContext): LogicalPlan = withOrigin(ctx) {//访问 queryNoWith节点val query = plan(ctx.queryNoWith)/*这个ctes不知道是干什么的,这条sql没有这个部分,我们可以查看sqlBase.g4关于ctes的定义ctes: WITH namedQuery (',' namedQuery)*;*/// Apply CTEsquery.optional(ctx.ctes) {val ctes = ctx.ctes.namedQuery.asScala.map { nCtx =>val namedQuery = visitNamedQuery(nCtx)(namedQuery.alias, namedQuery)}// Check for duplicate names.checkDuplicateKeys(ctes, ctx)With(query, ctes.toMap)}}

以此类推,我们就可以自己去追踪如何根据ast生成LogicalPlan

  /*** Create a logical plan for a regular (single-insert) query.* 创建一个logical计划,为一个常规的(简单查询)* 第3步*/override def visitSingleInsertQuery(ctx: SingleInsertQueryContext): LogicalPlan = withOrigin(ctx) {//访问queryTermDefault 没有实现,访问子节点plan(ctx.queryTerm).// Add organization statements.// 以with开头的方法添加信息到Logical planoptionalMap(ctx.queryOrganization)(withQueryResultClauses).// Add insert.optionalMap(ctx.insertInto())(withInsertInto)}/*** Create a logical plan using a query specification.* 用一个查询规范 创建一个 logical plan* 第4步  SELECT xxx from xxx where xxx having xxx  剔除了 sort by /order by /limit 等*/override def visitQuerySpecification(ctx: QuerySpecificationContext): LogicalPlan = withOrigin(ctx) {val from = OneRowRelation.optional(ctx.fromClause) {visitFromClause(ctx.fromClause)}withQuerySpecification(ctx, from)}

在了解了LogicalPlan,expression表达式后,再来继续.