Hudi Spark SQL源码学习总结-Create Table

2022-07-20

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

前言

简要总结Hudi Spark Sql源码执行逻辑，从建表开始。其实从去年开始接触Hudi的时候就研究学习了Hudi Spark SQL的部分源码，并贡献了几个PR，但是完整的逻辑有些地方还没有完全梳理清楚，所以现在想要从头开始学习，搞懂一些知识难点，这样以后看相关源码的时候就不会导致因为一些关键点不懂影响进度。

由于本人能力和精力有限，本文只讲解自己觉得比较关键的点，主要目的是梳理整个流程。

Spark SQL源码

既然是学习Hudi Spark SQL源码，那么肯定离不开Spark SQL源码，所以需要先学习了解Spark SQL的源码，在CSDN上发现一位作者写的几篇文章不错，这几天我也主要是参考他写的这几篇文章并结合源码进行学习的,我把它们放在后面的参考文章中，大家可以参考一下。

版本

Spark 2.4.4
Hudi master分支 0.12.0-SNAPSHOT

虽然在学习Spark SQL源码的时候用的是Spark3.3,但是因为Hudi源码默认的Spark版本是2.4.4，如果改版本在IDEA调试的话比较麻烦，所以是用Spark2.4.4版本，但我和Spark3.3对比了一下，大致逻辑是一样的。

示例代码

直接拿源码里的TestCreateTable的第一个测试语句

spark.sql(
      s"""
         | create table $tableName (
         |  id int,
         |  name string,
         |  price double,
         |  ts long
         | ) using hudi
         | tblproperties (
         |   hoodie.database.name = "databaseName",
         |   hoodie.table.name = "tableName",
         |   primaryKey = 'id',
         |   preCombineField = 'ts',
         |   hoodie.datasource.write.operation = 'upsert'
         | )
       """.stripMargin)

打印计划

我们先看一下上面的代码的执行计划

1 2	val df = spark.sql(createTableSql) df.explain(true)

打印结果

== Parsed Logical Plan ==
'CreateTable `h0`, ErrorIfExists

== Analyzed Logical Plan ==
CreateHoodieTableCommand `h0`, false

== Optimized Logical Plan ==
CreateHoodieTableCommand `h0`, false

== Physical Plan ==
Execute CreateHoodieTableCommand
   +- CreateHoodieTableCommand `h0`, false

分别对应阶段parsing、analysis、optimization 、planning

扩展 HoodieSparkSessionExtension

要想使Spark SQL支持Hudi,必须在启动spark的时候扩展 HoodieSparkSessionExtension,先放在这里，后面分析的时候会用到

package org.apache.spark.sql.hudi

import org.apache.hudi.SparkAdapterSupport
import org.apache.spark.sql.SparkSessionExtensions
import org.apache.spark.sql.hudi.analysis.HoodieAnalysis
import org.apache.spark.sql.parser.HoodieCommonSqlParser

/**
 * The Hoodie SparkSessionExtension for extending the syntax and add the rules.
 */
class HoodieSparkSessionExtension extends (SparkSessionExtensions => Unit)
  with SparkAdapterSupport {
  override def apply(extensions: SparkSessionExtensions): Unit = {

    extensions.injectParser { (session, parser) =>
      new HoodieCommonSqlParser(session, parser)
    }

    HoodieAnalysis.customResolutionRules.foreach { ruleBuilder =>
      extensions.injectResolutionRule { session =>
        ruleBuilder(session)
      }
    }

    HoodieAnalysis.customPostHocResolutionRules.foreach { rule =>
      extensions.injectPostHocResolutionRule { session =>
        rule(session)
      }
    }
  }
}

ANTLR

学习Spark SQL的时候我们知道SQL的识别、编译、解释是由ANTLR实现的，所以我们需要看一下Hudi源码里的.g4文件,一共有三个

hudi-spark模块下的 HoodieSqlCommon.g4
hudi-spark2模块下的 SqlBase.g4，这个其实是拷贝的Spark2.4.5源码里的SqlBase.g4
hudi-spark2模块下的 HoodieSqlBase.g4 其中导入了上面的SqlBase.g4:import SqlBase

入口

spark.sql

1
2
3

def sql(sqlText: String): DataFrame = {
  Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText))
}

sessionState.sqlParser

首先看一下sessionState是由哪个类实现的

lazy val sessionState: SessionState = {
  parentSessionState
    .map(_.clone(this))
    .getOrElse {
      val state = SparkSession.instantiateSessionState(
        SparkSession.sessionStateClassName(sparkContext.conf),
        self)
      initialSessionOptions.foreach { case (k, v) => state.conf.setConfString(k, v) }
      state
    }
}

private def sessionStateClassName(conf: SparkConf): String = {
  conf.get(CATALOG_IMPLEMENTATION) match {
    case "hive" => HIVE_SESSION_STATE_BUILDER_CLASS_NAME
    case "in-memory" => classOf[SessionStateBuilder].getCanonicalName
  }
}
private val HIVE_SESSION_STATE_BUILDER_CLASS_NAME =
  "org.apache.spark.sql.hive.HiveSessionStateBuilder"

  def enableHiveSupport(): Builder = synchronized {
    if (hiveClassesArePresent) {
      config(CATALOG_IMPLEMENTATION.key, "hive")
    } else {
      throw new IllegalArgumentException(
        "Unable to instantiate SparkSession with Hive support because " +
          "Hive classes are not found.")
    }
  }

可以看到在不启用Hive的情况下，使用的是SessionStateBuilder,启用了Hive使用HiveSessionStateBuilder,我们本地测试代码默认不启用Hive,那么看一下SessionStateBuilder

class SessionStateBuilder(
    session: SparkSession,
    parentState: Option[SessionState] = None)
  extends BaseSessionStateBuilder(session, parentState) {
  override protected def newBuilder: NewBuilder = new SessionStateBuilder(_, _)
}

SessionStateBuilder继承自BaseSessionStateBuilder,也就是sessionState是通过BaseSessionStateBuilder.build创建的，其中定义了sqlParser

1
2
3

protected lazy val sqlParser: ParserInterface = {
  extensions.buildParser(session, new SparkSqlParser(conf))
}

extensions 是在SparkSession.Builder中定义的：

1	private[this] val extensions = new SparkSessionExtensions

SparkSessionExtensions

type ParserBuilder = (SparkSession, ParserInterface) => ParserInterface
private[this] val parserBuilders = mutable.Buffer.empty[ParserBuilder]
private[sql] def buildParser(
    session: SparkSession,
    initial: ParserInterface): ParserInterface = {
  parserBuilders.foldLeft(initial) { (parser, builder) =>
    builder(session, parser)
  }
}

默认情况下parserBuilders为空，那么foldLeft返回的值为默认的SparkSqlParser,一开始没搞懂parserBuilders为空的情况下为啥会返回SparkSqlParser，直到看了foldLeft的源码：

def foldLeft[B](z: B)(op: (B, A) => B): B = {
  var result = z
  this foreach (x => result = op(result, x))
  result
}

首先将参数z赋值为result，然后再foreach利用op函数改变result的值，因为这里parserBuilders为空，所以没有执行foreach里面的逻辑，那么返回结果result为参数z，也就是这里的new SparkSqlParser(conf)

Hudi自定义sqlParser

搞懂了Spark SQL默认的sqlParser为SparkSqlParser,那么Hudi是一样的吗？那我们就需要看一下开始的HoodieSparkSessionExtension,我们在程序里是这样扩展的.withExtensions(new HoodieSparkSessionExtension)

def withExtensions(f: SparkSessionExtensions => Unit): Builder = synchronized {
  f(extensions)
  this
}

这里的f为HoodieSparkSessionExtension（extends (SparkSessionExtensions => Unit)，继承Function1 ），也就是一开始withExtensions时会调用HoodieSparkSessionExtension的apply方法，其中会执行

1
2
3

extensions.injectParser { (session, parser) =>
  new HoodieCommonSqlParser(session, parser)
}

这里的injectParser会将HoodieCommonSqlParser添加到parserBuilders中

1
2
3

def injectParser(builder: ParserBuilder): Unit = {
  parserBuilders += builder
}

再根据上面对buildParser的理解，那么在Hudi Spark SQL中sqlParser为HoodieCommonSqlParser

parsing

parsePlan

接下来看parsePlan方法，这个方法对应parsing阶段，这里实际为HoodieCommonSqlParser.parsePlan

private lazy val builder = new HoodieSqlCommonAstBuilder(session, delegate)
private lazy val sparkExtendedParser = sparkAdapter.createExtendedSparkParser
  .map(_(session, delegate)).getOrElse(delegate)

override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
  builder.visit(parser.singleStatement()) match {
    case plan: LogicalPlan => plan
    case _=> sparkExtendedParser.parsePlan(sqlText)
  }
}

先看parse方法

  protected def parse[T](command: String)(toResult: HoodieSqlCommonParser => T): T = {
    logDebug(s"Parsing command: $command")

    val lexer = new HoodieSqlCommonLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
    lexer.removeErrorListeners()
    lexer.addErrorListener(ParseErrorListener)

    val tokenStream = new CommonTokenStream(lexer)
    val parser = new HoodieSqlCommonParser(tokenStream)
    parser.removeErrorListeners()
    parser.addErrorListener(ParseErrorListener)

    try {
      try {
        // first, try parsing with potentially faster SLL mode
        parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
        toResult(parser)
      }
      catch {
        case e: ParseCancellationException =>
          // if we fail, parse with LL mode
          tokenStream.seek(0) // rewind input stream
          parser.reset()

          // Try Again.
          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)
    }
  }
}

根据上面的代码可知函数体中的parser为HoodieSqlCommonParser，这里的HoodieSqlCommonParser为HoodieSqlCommon.g4编译生成的，根据在参考文章中学习Spark SQL源码时可知parser.singleStatement()的核心代码是调用方法statement，而statement根据我的理解对应g4文件中的statement，根据g4文件中定义的语句匹配我们传入的sql语句，如果匹配上返回对应的LogicalPlan，匹配不成功则返回null

   public final SingleStatementContext singleStatement() throws RecognitionException {
	SingleStatementContext _localctx = new SingleStatementContext(_ctx, getState());
	enterRule(_localctx, 0, RULE_singleStatement);
	int _la;
	try {
		enterOuterAlt(_localctx, 1);
		{
		setState(40);
		statement();
     ......

public final StatementContext statement() throws RecognitionException {
	StatementContext _localctx = new StatementContext(_ctx, getState());
	enterRule(_localctx, 2, RULE_statement);
	int _la;
	try {
		int _alt;
		setState(124);
		_errHandler.sync(this);
		switch ( getInterpreter().adaptivePredict(_input,12,_ctx) ) {
		case 1:  
               ....

而HoodieSqlCommon.g4文件中的statement没有对应create table相关的语法:

singleStatement
   : statement ';'* EOF
   ;

statement
   : compactionStatement                                                       #compactionCommand
   | CALL multipartIdentifier '(' (callArgument (',' callArgument)*)? ')'      #call
   | CREATE INDEX (IF NOT EXISTS)? identifier ON TABLE?
         tableIdentifier (USING indexType=identifier)?
         LEFT_PAREN columns=multipartIdentifierPropertyList RIGHT_PAREN
         (OPTIONS indexOptions=propertyList)?                                  #createIndex
   | DROP INDEX (IF EXISTS)? identifier ON TABLE? tableIdentifier              #dropIndex
   | SHOW INDEXES (FROM | IN) TABLE? tableIdentifier                           #showIndexes
   | REFRESH INDEX identifier ON TABLE? tableIdentifier                        #refreshIndex
   | .*?                                                                       #passThrough
   ;

可以看出HoodieSqlCommonParser的singleStatement的实现也是和g4文件中的定义一致

所以在这里parser.singleStatement()返回null,那么builder.visit(parser.singleStatement())模式匹配会走到sparkExtendedParser.parsePlan(sqlText)

sparkExtendedParser

private lazy val sparkExtendedParser = sparkAdapter.createExtendedSparkParser
  .map(_(session, delegate)).getOrElse(delegate)

lazy val sparkAdapter: SparkAdapter = {
  val adapterClass = if (HoodieSparkUtils.isSpark3_2) {
    "org.apache.spark.sql.adapter.Spark3_2Adapter"
  } else if (HoodieSparkUtils.isSpark3_0 || HoodieSparkUtils.isSpark3_1) {
    "org.apache.spark.sql.adapter.Spark3_1Adapter"
  } else {
    "org.apache.spark.sql.adapter.Spark2Adapter"
  }
  getClass.getClassLoader.loadClass(adapterClass)
    .newInstance().asInstanceOf[SparkAdapter]
}

Spark2Adapter.createExtendedSparkParser

override def createExtendedSparkParser: Option[(SparkSession, ParserInterface) => ParserInterface] = {
  Some(
    (spark: SparkSession, delegate: ParserInterface) => new HoodieSpark2ExtendedSqlParser(spark, delegate)
  )
}

HoodieSpark2ExtendedSqlParser.parsePlan

override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
  builder.visit(parser.singleStatement()) match {
    case plan: LogicalPlan => plan
    case _=> delegate.parsePlan(sqlText)
  }
}

这里的parsePlan和上面的差不多，还是要先看parse方法


protected def parse[T](command: String)(toResult: HoodieSqlBaseParser => T): T = {
  logDebug(s"Parsing command: $command")

  val lexer = new HoodieSqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
  lexer.removeErrorListeners()
  lexer.addErrorListener(ParseErrorListener)
  lexer.legacy_setops_precedence_enbled = conf.setOpsPrecedenceEnforced

  val tokenStream = new CommonTokenStream(lexer)
  val parser = new HoodieSqlBaseParser(tokenStream)
  parser.addParseListener(PostProcessor)
  parser.removeErrorListeners()
  parser.addErrorListener(ParseErrorListener)
  parser.legacy_setops_precedence_enbled = conf.setOpsPrecedenceEnforced

  try {
    try {
      // first, try parsing with potentially faster SLL mode
      parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
      toResult(parser)
    }
    catch {
      case e: ParseCancellationException =>
        // if we fail, parse with LL mode
        tokenStream.seek(0) // rewind input stream
        parser.reset()

        // Try Again.
        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)
  }
}

这里的parser为HoodieSqlBaseParser，对应HoodieSqlBase.g4

singleStatement
    : statement EOF
    ;

statement
    : mergeInto                                                        #mergeIntoTable
    | updateTableStmt                                                  #updateTable
    | deleteTableStmt                                                  #deleteTable
    | .*?                                                              #passThrough
    ;

同样的逻辑这里的statement也没有create table,同样返回null，调用delegate.parsePlan(sqlText),这里的delegate由最开始的extensions.buildParser传进来的

1	extensions.buildParser(session, new SparkSqlParser(conf))

所以这里的delegate为SparkSqlParser

SparkSqlParser

SparkSqlParser为Spark的源码，继承了抽象类AbstractSqlParser,本身并没有实现parsePlan方法，所以需要看一下父类AbstractSqlParser的parsePlan方法

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)
  }
}

同样的先看parse方法

  protected def parse[T](command: String)(toResult: SqlBaseParser => T): T = {
    logDebug(s"Parsing command: $command")

    val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
    lexer.removeErrorListeners()
    lexer.addErrorListener(ParseErrorListener)
    lexer.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced

    val tokenStream = new CommonTokenStream(lexer)
    val parser = new SqlBaseParser(tokenStream)
    parser.addParseListener(PostProcessor)
    parser.removeErrorListeners()
    parser.addErrorListener(ParseErrorListener)
    parser.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced

    try {
      try {
        // first, try parsing with potentially faster SLL mode
        parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
        toResult(parser)
      }
      catch {
        case e: ParseCancellationException =>
          // if we fail, parse with LL mode
          tokenStream.seek(0) // rewind input stream
          parser.reset()

          // Try Again.
          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)
    }
  }
}

这里的parser为SqlBaseParser,它是Spark源码里的,同样的Spark源码里也有一个SqlBase.g4，其实Hudi的SqlBase.g4文件是拷贝自Spark源码里的，对比两个文件的内容和从Hudi的SqlBase.g4里的一行注释就可以知晓：

1	// The parser file is forked from spark 2.4.5's SqlBase.g4.

我们在SqlBase.g4里查看singleStatement和statement的定义

singleStatement
    : statement EOF
    ;

statement
    : query                                                            #statementDefault
    | USE db=identifier                                                #use
    | CREATE DATABASE (IF NOT EXISTS)? identifier
        (COMMENT comment=STRING)? locationSpec?
        (WITH DBPROPERTIES tablePropertyList)?                         #createDatabase
    | ALTER DATABASE identifier SET DBPROPERTIES tablePropertyList     #setDatabaseProperties
    | DROP DATABASE (IF EXISTS)? identifier (RESTRICT | CASCADE)?      #dropDatabase
    | createTableHeader ('(' colTypeList ')')? tableProvider
        ((OPTIONS options=tablePropertyList) |
        (PARTITIONED BY partitionColumnNames=identifierList) |
        bucketSpec |
        locationSpec |
        (COMMENT comment=STRING) |
        (TBLPROPERTIES tableProps=tablePropertyList))*
        (AS? query)?                                                   #createTable
    | createTableHeader ('(' columns=colTypeList ')')?
        ((COMMENT comment=STRING) |
        (PARTITIONED BY '(' partitionColumns=colTypeList ')') |
        bucketSpec |
        skewSpec |
        rowFormat |
        createFileFormat |
        locationSpec |
        (TBLPROPERTIES tableProps=tablePropertyList))*
        (AS? query)?                                                   #createHiveTable
        ......

tableProvider
    : USING qualifiedName
    ;

statement内容比较多，只截取其中一部分，可以看到statement有两个和sql create table相关的，一个是#createTable另一个是#createHiveTable,他俩的区别为#createTable多了一个tableProvider,tableProvider 包含关键字USING，我们的sql里有using hudi,所以这里的statement应该匹配为#createTable,具体看一下它的部分源码

singleStatement和上面的差不多，核心逻辑还是statement()

public final SingleStatementContext singleStatement() throws RecognitionException {
	SingleStatementContext _localctx = new SingleStatementContext(_ctx, getState());
	enterRule(_localctx, 0, RULE_singleStatement);
	try {
		enterOuterAlt(_localctx, 1);
		{
		setState(202);
		statement();
		setState(203);
		match(EOF);
		}
	}
	catch (RecognitionException re) {
		_localctx.exception = re;
		_errHandler.reportError(this, re);
		_errHandler.recover(this, re);
	}
	finally {
		exitRule();
	}
	return _localctx;
}

statement

public final StatementContext statement() throws RecognitionException {
		StatementContext _localctx = new StatementContext(_ctx, getState());
		enterRule(_localctx, 12, RULE_statement);
		int _la;
		try {
			int _alt;
			setState(826);
			_errHandler.sync(this);
			switch ( getInterpreter().adaptivePredict(_input,95,_ctx) ) {
			case 1:
        _localctx = new StatementDefaultContext(_localctx);
        ......
			case 6:
				_localctx = new CreateTableContext(_localctx);
				enterOuterAlt(_localctx, 6);
				{
				setState(260);
				createTableHeader();
				setState(265);
				_errHandler.sync(this);
				_la = _input.LA(1);
				if (_la==T__0) {
					{
					setState(261);
					match(T__0);
					setState(262);
					colTypeList();
					setState(263);
					match(T__1);
					}
				}

				setState(267);
				tableProvider();
				setState(281);
				_errHandler.sync(this);
				_la = _input.LA(1);
				while (_la==COMMENT || ((((_la - 183)) & ~0x3f) == 0 && ((1L << (_la - 183)) & ((1L << (OPTIONS - 183)) | (1L << (TBLPROPERTIES - 183)) | (1L << (LOCATION - 183)) | (1L << (CLUSTERED - 183)) | (1L << (PARTITIONED - 183)))) != 0)) {
					{
					setState(279);
					_errHandler.sync(this);
					switch (_input.LA(1)) {
					case OPTIONS:
						{
						{
						setState(268);
						match(OPTIONS);
						setState(269);
						((CreateTableContext)_localctx).options = tablePropertyList();
						}
						}
						break;
					case PARTITIONED:
						{
						{
						setState(270);
						match(PARTITIONED);
						setState(271);
						match(BY);
						setState(272);
						((CreateTableContext)_localctx).partitionColumnNames = identifierList();
						}
						}
						break;
					case CLUSTERED:
						{
						setState(273);
						bucketSpec();
						}
						break;
					case LOCATION:
						{
						setState(274);
						locationSpec();
						}
						break;
					case COMMENT:
						{
						{
						setState(275);
						match(COMMENT);
						setState(276);
						((CreateTableContext)_localctx).comment = match(STRING);
						}
						}
						break;
					case TBLPROPERTIES:
						{
						{
						setState(277);
						match(TBLPROPERTIES);
						setState(278);
						((CreateTableContext)_localctx).tableProps = tablePropertyList();
						}
						}
						break;
					default:
						throw new NoViableAltException(this);
					}
					}
					setState(283);
					_errHandler.sync(this);
					_la = _input.LA(1);
				}
				setState(288);
				_errHandler.sync(this);
				_la = _input.LA(1);
				if ((((_la) & ~0x3f) == 0 && ((1L << _la) & ((1L << T__0) | (1L << SELECT) | (1L << FROM) | (1L << AS))) != 0) || ((((_la - 77)) & ~0x3f) == 0 && ((1L << (_la - 77)) & ((1L << (WITH - 77)) | (1L << (VALUES - 77)) | (1L << (TABLE - 77)) | (1L << (INSERT - 77)) | (1L << (MAP - 77)))) != 0) || _la==REDUCE) {
					{
					setState(285);
					_errHandler.sync(this);
					_la = _input.LA(1);
					if (_la==AS) {
						{
						setState(284);
						match(AS);
						}
					}

					setState(287);
					query();
					}
				}

				}
				break;
			case 7:
				_localctx = new CreateHiveTableContext(_localctx);
				enterOuterAlt(_localctx, 7);
				{
				setState(290);
				createTableHeader();
				setState(295);

这里应该匹配到 6 ：CreateTableContext,然后去匹配g4文件中定义的 createTableHeader、tableProvider、PARTITIONED、TBLPROPERTIES等,主要是匹配sql关键字解析语法

解释一下为啥会匹配到6 ：CreateTableContext，我没有专门去学习ANTLR的原理，根据我的理解，statement定义了很多语法，其中#createTable是第6个，所以会匹配到6，至于CreateTableContext,是将#createTable首字母大写并在后面加个Context，里面的其他内容也是和g4文件中的定义一一对应的，可以看一下其他的匹配都是这个规律

parser.singleStatement() 其实返回的是SingleStatementContext,接下来看一下visitSingleStatement，先看一下参数ctx.statement

 override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan = withOrigin(ctx) {
   visit(ctx.statement).asInstanceOf[LogicalPlan]
 }

 public static class SingleStatementContext extends ParserRuleContext {
	public StatementContext statement() {
		return getRuleContext(StatementContext.class,0);
	}
       ......

   // 这里的参数i等于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) {
       // 如果Statement的子节点为null或者i>0子节点的大小
	if ( children==null || i < 0 || i >= children.size() ) {
		return null;
	}

	int j = -1; // what element have we found with ctxType?
       // 遍历children
	for (ParseTree o : children) {
           // 如果o是StatementContext的子类
           // 这里的第一个子节点为CreateTableContext，它是StatementContext
		if ( ctxType.isInstance(o) ) {
			j++; // j=0
			if ( j == i ) { //相等返回 CreateTableContext
				return ctxType.cast(o);
			}
		}
	}
	return null;
}

根据上面的分析和我在代码中的注释我们知道这里的ctx.statement返回的是CreateTableContext,再看一下visit

1
2
3

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

这里的tree为CreateTableContext，所以实际调用CreateTableContext的accept方法

public static class CreateTableContext extends StatementContext {
	public <T> T accept(ParseTreeVisitor<? extends T> visitor) {
		if ( visitor instanceof SqlBaseVisitor ) return ((SqlBaseVisitor<? extends T>)visitor).visitCreateTable(this);
		else return visitor.visitChildren(this);
	}

accept又调用visitCreateTable,这里的visitor为前面讲到的SparkSqlParser里面的SparkSqlAstBuilder ,它是AstBuilder的子类，AstBuilder是SqlBaseBaseVisitor的子类

1 2	class SparkSqlAstBuilder extends AstBuilder class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logging

override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = withOrigin(ctx) {
  val (table, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
  if (external) {
    operationNotAllowed("CREATE EXTERNAL TABLE ... USING", ctx)
  }

  checkDuplicateClauses(ctx.TBLPROPERTIES, "TBLPROPERTIES", ctx)
  checkDuplicateClauses(ctx.OPTIONS, "OPTIONS", ctx)
  checkDuplicateClauses(ctx.PARTITIONED, "PARTITIONED BY", ctx)
  checkDuplicateClauses(ctx.COMMENT, "COMMENT", ctx)
  checkDuplicateClauses(ctx.bucketSpec(), "CLUSTERED BY", ctx)
  checkDuplicateClauses(ctx.locationSpec, "LOCATION", ctx)

  val options = Option(ctx.options).map(visitPropertyKeyValues).getOrElse(Map.empty)
  // provider为hudi
  val provider = ctx.tableProvider.qualifiedName.getText
  val schema = Option(ctx.colTypeList()).map(createSchema)
  val partitionColumnNames =
    Option(ctx.partitionColumnNames)
      .map(visitIdentifierList(_).toArray)
      .getOrElse(Array.empty[String])
  val properties = Option(ctx.tableProps).map(visitPropertyKeyValues).getOrElse(Map.empty)
  val bucketSpec = ctx.bucketSpec().asScala.headOption.map(visitBucketSpec)

  val location = ctx.locationSpec.asScala.headOption.map(visitLocationSpec)
  val storage = DataSource.buildStorageFormatFromOptions(options)

  if (location.isDefined && storage.locationUri.isDefined) {
    throw new ParseException(
      "LOCATION and 'path' in OPTIONS are both used to indicate the custom table path, " +
        "you can only specify one of them.", ctx)
  }
  val customLocation = storage.locationUri.orElse(location.map(CatalogUtils.stringToURI))

  val tableType = if (customLocation.isDefined) {
    CatalogTableType.EXTERNAL
  } else {
    CatalogTableType.MANAGED
  }

  val tableDesc = CatalogTable(
    identifier = table,
    tableType = tableType,
    storage = storage.copy(locationUri = customLocation),
    schema = schema.getOrElse(new StructType),
    provider = Some(provider),
    partitionColumnNames = partitionColumnNames,
    bucketSpec = bucketSpec,
    properties = properties,
    comment = Option(ctx.comment).map(string))

  // Determine the storage mode.
  val mode = if (ifNotExists) SaveMode.Ignore else SaveMode.ErrorIfExists

  if (ctx.query != null) {
    // Get the backing query.
    val query = plan(ctx.query)

    if (temp) {
      operationNotAllowed("CREATE TEMPORARY TABLE ... USING ... AS query", ctx)
    }

    // Don't allow explicit specification of schema for CTAS
    if (schema.nonEmpty) {
      operationNotAllowed(
        "Schema may not be specified in a Create Table As Select (CTAS) statement",
        ctx)
    }
    CreateTable(tableDesc, mode, Some(query))
  } else {
    if (temp) {
      if (ifNotExists) {
        operationNotAllowed("CREATE TEMPORARY TABLE IF NOT EXISTS", ctx)
      }

      logWarning(s"CREATE TEMPORARY TABLE ... USING ... is deprecated, please use " +
        "CREATE TEMPORARY VIEW ... USING ... instead")
      // Unlike CREATE TEMPORARY VIEW USING, CREATE TEMPORARY TABLE USING does not support
      // IF NOT EXISTS. Users are not allowed to replace the existing temp table.
      CreateTempViewUsing(table, schema, replace = false, global = false, provider, options)
    } else {
      CreateTable(tableDesc, mode, None)
    }
  }
}

具体细节就不多做解释，可以自己看，主要的点一个是provider为hudi这个后面会用到，它是在前面提到的tableProvider得到的，返回结果为CreateTable，它是LogicalPlan的子类

case class CreateTable(
    tableDesc: CatalogTable,
    mode: SaveMode,
    query: Option[LogicalPlan]) extends LogicalPlan {
  assert(tableDesc.provider.isDefined, "The table to be created must have a provider.")

  if (query.isEmpty) {
    assert(
      mode == SaveMode.ErrorIfExists || mode == SaveMode.Ignore,
      "create table without data insertion can only use ErrorIfExists or Ignore as SaveMode.")
  }

  override def children: Seq[LogicalPlan] = query.toSeq
  override def output: Seq[Attribute] = Seq.empty
  override lazy val resolved: Boolean = false
}

这样parsePlan最终返回的结果为 CreateTable,parsing阶段完成，接下来到了analysis阶段

analysis

def sql(sqlText: String): DataFrame = {
   Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText))
 }

 def ofRows(sparkSession: SparkSession, logicalPlan: LogicalPlan): DataFrame = {
   val qe = sparkSession.sessionState.executePlan(logicalPlan)
   qe.assertAnalyzed()
   new Dataset[Row](sparkSession, qe, RowEncoder(qe.analyzed.schema))
 }

analysis阶段的入口在qe.assertAnalyzed()

def assertAnalyzed(): Unit = analyzed

lazy val analyzed: LogicalPlan = {
  SparkSession.setActiveSession(sparkSession)
  sparkSession.sessionState.analyzer.executeAndCheck(logical)
}

analyzer

首先看一下sparkSession.sessionState.analyzer

1	lazy val analyzer: Analyzer = analyzerBuilder()

上面提到过sessionState是由BaseSessionStateBuilder.build创建的，那么同样去BaseSessionStateBuilder里面看一下这个analyzerBuilder是啥

protected def analyzer: Analyzer = new Analyzer(catalog, conf) {
  override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
    new FindDataSourceTable(session) +:
      new ResolveSQLOnFile(session) +:
      customResolutionRules

  override val postHocResolutionRules: Seq[Rule[LogicalPlan]] =
    PreprocessTableCreation(session) +:
      PreprocessTableInsertion(conf) +:
      DataSourceAnalysis(conf) +:
      customPostHocResolutionRules

  override val extendedCheckRules: Seq[LogicalPlan => Unit] =
    PreWriteCheck +:
      PreReadCheck +:
      HiveOnlyCheck +:
      customCheckRules
}

它是一个匿名内部类重写了Analyzer的几个扩展规则，这几个扩展规则在Analyzer中均为空，其中extendedCheckRules是在Analyzer的父接口CheckAnalysis中定义的，我们上面提到了，Hudi第一个自定义扩展是扩展了自己的sqlParser,那么剩下的两个都是作用在analysis阶段

HoodieAnalysis.customResolutionRules.foreach { ruleBuilder =>
  extensions.injectResolutionRule { session =>
    ruleBuilder(session)
  }
}

HoodieAnalysis.customPostHocResolutionRules.foreach { rule =>
  extensions.injectPostHocResolutionRule { session =>
    rule(session)
  }
}

injectResolutionRule

先看一下injectResolutionRule

1
2
3

def injectResolutionRule(builder: RuleBuilder): Unit = {
  resolutionRuleBuilders += builder
}

结合BaseSessionStateBuilder中的customResolutionRules,可知，这第二个自定义扩展是实现了extendedResolutionRules中的customResolutionRules,对应Analyzer的规则批batches的 Resolution

protected def customResolutionRules: Seq[Rule[LogicalPlan]] = {
  extensions.buildResolutionRules(session)
}

private[sql] def buildResolutionRules(session: SparkSession): Seq[Rule[LogicalPlan]] = {
  resolutionRuleBuilders.map(_.apply(session))
}

injectPostHocResolutionRule

再看一下injectPostHocResolutionRule

def injectPostHocResolutionRule(builder: RuleBuilder): Unit = {
  postHocResolutionRuleBuilders += builder
}

protected def customPostHocResolutionRules: Seq[Rule[LogicalPlan]] = {
  extensions.buildPostHocResolutionRules(session)
}

private[sql] def buildPostHocResolutionRules(session: SparkSession): Seq[Rule[LogicalPlan]] = {
  postHocResolutionRuleBuilders.map(_.apply(session))
}

可知第三个自定义扩展是实现了postHocResolutionRules中的customPostHocResolutionRules,对应Analyzer的规则批batches的 Post-Hoc Resolution

Hudi自定义analysis规则

Hudi自定义扩展具体的规则有

extendedResolutionRules: session => HoodieResolveReferences(session) 和 session => HoodieAnalysis(session)

postHocResolutionRules: session => HoodiePostAnalysisRule(session)

具体代码为：

def customResolutionRules: Seq[RuleBuilder] = {
  val rules: ListBuffer[RuleBuilder] = ListBuffer(
    // Default rules
    session => HoodieResolveReferences(session),
    session => HoodieAnalysis(session)
  )

  if (HoodieSparkUtils.gteqSpark3_2) {
    val dataSourceV2ToV1FallbackClass = "org.apache.spark.sql.hudi.analysis.HoodieDataSourceV2ToV1Fallback"
    val dataSourceV2ToV1Fallback: RuleBuilder =
      session => ReflectionUtils.loadClass(dataSourceV2ToV1FallbackClass, session).asInstanceOf[Rule[LogicalPlan]]

    val spark3AnalysisClass = "org.apache.spark.sql.hudi.analysis.HoodieSpark3Analysis"
    val spark3Analysis: RuleBuilder =
      session => ReflectionUtils.loadClass(spark3AnalysisClass, session).asInstanceOf[Rule[LogicalPlan]]

    val spark3ResolveReferencesClass = "org.apache.spark.sql.hudi.analysis.HoodieSpark3ResolveReferences"
    val spark3ResolveReferences: RuleBuilder =
      session => ReflectionUtils.loadClass(spark3ResolveReferencesClass, session).asInstanceOf[Rule[LogicalPlan]]

    val spark32ResolveAlterTableCommandsClass = "org.apache.spark.sql.hudi.ResolveHudiAlterTableCommandSpark32"
    val spark32ResolveAlterTableCommands: RuleBuilder =
      session => ReflectionUtils.loadClass(spark32ResolveAlterTableCommandsClass, session).asInstanceOf[Rule[LogicalPlan]]

    // NOTE: PLEASE READ CAREFULLY
    //
    // It's critical for this rules to follow in this order, so that DataSource V2 to V1 fallback
    // is performed prior to other rules being evaluated
    rules ++= Seq(dataSourceV2ToV1Fallback, spark3Analysis, spark3ResolveReferences, spark32ResolveAlterTableCommands)

  } else if (HoodieSparkUtils.gteqSpark3_1) {
    val spark31ResolveAlterTableCommandsClass = "org.apache.spark.sql.hudi.ResolveHudiAlterTableCommand312"
    val spark31ResolveAlterTableCommands: RuleBuilder =
      session => ReflectionUtils.loadClass(spark31ResolveAlterTableCommandsClass, session).asInstanceOf[Rule[LogicalPlan]]

    rules ++= Seq(spark31ResolveAlterTableCommands)
  }

  rules
}

def customPostHocResolutionRules: Seq[RuleBuilder] = {
  val rules: ListBuffer[RuleBuilder] = ListBuffer(
    // Default rules
    session => HoodiePostAnalysisRule(session)
  )

  if (HoodieSparkUtils.gteqSpark3_2) {
    val spark3PostHocResolutionClass = "org.apache.spark.sql.hudi.analysis.HoodieSpark3PostAnalysisRule"
    val spark3PostHocResolution: RuleBuilder =
      session => ReflectionUtils.loadClass(spark3PostHocResolutionClass, session).asInstanceOf[Rule[LogicalPlan]]

    rules += spark3PostHocResolution
  }

  rules
}

可以看出来如果是spark3的话，还有别的规则，因为这里是spark2,所以只看默认的规则就可以了

看完sparkSession.sessionState.analyzer的定义，接下来就看analysis具体的实现方法了

executeAndCheck

def executeAndCheck(plan: LogicalPlan): LogicalPlan = AnalysisHelper.markInAnalyzer {
  val analyzed = execute(plan)
  try {
    checkAnalysis(analyzed)
    analyzed
  } catch {
    case e: AnalysisException =>
      val ae = new AnalysisException(e.message, e.line, e.startPosition, Option(analyzed))
      ae.setStackTrace(e.getStackTrace)
      throw ae
  }
}

根据参考文章中的学习，我们知道，主要分为2 步：
1、执行 execute(plan)
2、校验 checkAnalysis(analyzed)

这里主要讲Hudi的实现，我们知道在执行阶段实际是调用父类RuleExecutor的execute方法，它会遍历规则批中的规则，并应用规则，调用规则的apply方法

override def execute(plan: LogicalPlan): LogicalPlan = {
  AnalysisContext.reset()
  try {
    executeSameContext(plan)
  } finally {
    AnalysisContext.reset()
  }
}

private def executeSameContext(plan: LogicalPlan): LogicalPlan = super.execute(plan)

def execute(plan: TreeType): TreeType = {
    var curPlan = plan
    val queryExecutionMetrics = RuleExecutor.queryExecutionMeter

    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()
            // 调用规则的`apply`方法
            val result = rule(plan)
            val runTime = System.nanoTime() - startTime

            if (!result.fastEquals(plan)) {
              queryExecutionMetrics.incNumEffectiveExecution(rule.ruleName)
              queryExecutionMetrics.incTimeEffectiveExecutionBy(rule.ruleName, runTime)
              logTrace(
                s"""
                  |=== Applying Rule ${rule.ruleName} ===
                  |${sideBySide(plan.treeString, result.treeString).mkString("\n")}
                """.stripMargin)
            }

应用Hudi规则

规则批中有很多规则，这里只跟踪规则如何作用到Hudi的自定义规则中，并最终执行Hudi的建表的，首先遍历到extendedResolutionRules的customResolutionRules,具体有两个

第一个规则：

case class HoodieResolveReferences(sparkSession: SparkSession) extends Rule[LogicalPlan]
  with SparkAdapterSupport {
  private lazy val analyzer = sparkSession.sessionState.analyzer

  def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperatorsUp {
    // Resolve merge into
    case mergeInto @ MergeIntoTable(target, source, mergeCondition, matchedActions, notMatchedActions)
      if sparkAdapter.isHoodieTable(target, sparkSession) && target.resolved =>
      val resolver = sparkSession.sessionState.conf.resolver
      val resolvedSource = analyzer.execute(source)

      def isInsertOrUpdateStar(assignments: Seq[Assignment]): Boolean = {
        if (assignments.isEmpty) {
          true
        } else {
            ......

因为这里为create table，所以没有匹配到 mergeInto等

第二个规则：

override def apply(plan: LogicalPlan): LogicalPlan = {
  plan match {
    // Convert to MergeIntoHoodieTableCommand
    case m @ MergeIntoTable(target, _, _, _, _)
      if m.resolved && sparkAdapter.isHoodieTable(target, sparkSession) =>
        MergeIntoHoodieTableCommand(m)

    // Convert to UpdateHoodieTableCommand
    case u @ UpdateTable(table, _, _)
      if u.resolved && sparkAdapter.isHoodieTable(table, sparkSession) =>
        UpdateHoodieTableCommand(u)

    // Convert to DeleteHoodieTableCommand
    case d @ DeleteFromTable(table, _)
      if d.resolved && sparkAdapter.isHoodieTable(table, sparkSession) =>
        DeleteHoodieTableCommand(d)

    // Convert to InsertIntoHoodieTableCommand
    case l if sparkAdapter.isInsertInto(l) =>
      val (table, partition, query, overwrite, _) = sparkAdapter.getInsertIntoChildren(l).get
      table match {
        case relation: LogicalRelation if sparkAdapter.isHoodieTable(relation, sparkSession) =>
          new InsertIntoHoodieTableCommand(relation, query, partition, overwrite)
        case _ =>
          l
      }

    // Convert to CreateHoodieTableAsSelectCommand
    case CreateTable(table, mode, Some(query))
      if query.resolved && sparkAdapter.isHoodieTable(table) =>
        CreateHoodieTableAsSelectCommand(table, mode, query)

    // Convert to CompactionHoodieTableCommand
    case CompactionTable(table, operation, options)
      if table.resolved && sparkAdapter.isHoodieTable(table, sparkSession) =>
      val tableId = getTableIdentifier(table)
      val catalogTable = sparkSession.sessionState.catalog.getTableMetadata(tableId)
      CompactionHoodieTableCommand(catalogTable, operation, options)
    // Convert to CompactionHoodiePathCommand
    case CompactionPath(path, operation, options) =>

虽然这里为CreateTable，但是query为空，所以没有匹配到，不会走到CreateHoodieTableAsSelectCommand,也就是customResolutionRules的两个规则都没有起到实际作用，但是如果是其他的语句，比如merge into、ctas等到是会起作用的。

接下来遍历postHocResolutionRules,其中有两个规则比较重要DataSourceAnalysis(conf)和customPostHocResolutionRules,首先应用规则DataSourceAnalysis

case class DataSourceAnalysis(conf: SQLConf) extends Rule[LogicalPlan] with CastSupport {

    override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
    case CreateTable(tableDesc, mode, None) if DDLUtils.isDatasourceTable(tableDesc) =>
      DDLUtils.checkDataColNames(tableDesc)
      CreateDataSourceTableCommand(tableDesc, ignoreIfExists = mode == SaveMode.Ignore)

    case CreateTable(tableDesc, mode, Some(query))
        if query.resolved && DDLUtils.isDatasourceTable(tableDesc) =>
      DDLUtils.checkDataColNames(tableDesc.copy(schema = query.schema))
      CreateDataSourceTableAsSelectCommand(tableDesc, mode, query, query.output.map(_.name))

    case InsertIntoTable(l @ LogicalRelation(_: InsertableRelation, _, _, _),
        parts, query, overwrite, false) if parts.isEmpty =>
      InsertIntoDataSourceCommand(l, query, overwrite)

    case InsertIntoDir(_, storage, provider, query, overwrite)
      if provider.isDefined && provider.get.toLowerCase(Locale.ROOT) != DDLUtils.HIVE_PROVIDER =>

      val outputPath = new Path(storage.locationUri.get)
      if (overwrite) DDLUtils.verifyNotReadPath(query, outputPath)

      InsertIntoDataSourceDirCommand(storage, provider.get, query, overwrite)
      ......

   def isDatasourceTable(table: CatalogTable): Boolean = {
    table.provider.isDefined && table.provider.get.toLowerCase(Locale.ROOT) != HIVE_PROVIDER
  }

    val HIVE_PROVIDER = "hive"

这里会匹配到第一个case,根据前面讲的我们知道它是CreateTable(tableDesc, mode, None),它的provider为hudi，所以是datasourceTable,也就是经过这个规则应用后CreateTable转变为了CreateDataSourceTableCommand

再看规则customPostHocResolutionRules

case class HoodiePostAnalysisRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {
  override def apply(plan: LogicalPlan): LogicalPlan = {
    plan match {
      // Rewrite the CreateDataSourceTableCommand to CreateHoodieTableCommand
      case CreateDataSourceTableCommand(table, ignoreIfExists)
        if sparkAdapter.isHoodieTable(table) =>
        CreateHoodieTableCommand(table, ignoreIfExists)
      // Rewrite the DropTableCommand to DropHoodieTableCommand
      case DropTableCommand(tableName, ifExists, isView, purge)
        if sparkAdapter.isHoodieTable(tableName, sparkSession) =>
        DropHoodieTableCommand(tableName, ifExists, isView, purge)
      // Rewrite the AlterTableDropPartitionCommand to AlterHoodieTableDropPartitionCommand
      case AlterTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData)
        if sparkAdapter.isHoodieTable(tableName, sparkSession) =>
          AlterHoodieTableDropPartitionCommand(tableName, specs, ifExists, purge, retainData)
      // Rewrite the AlterTableRenameCommand to AlterHoodieTableRenameCommand
      // Rewrite the AlterTableAddColumnsCommand to AlterHoodieTableAddColumnsCommand
      case AlterTableAddColumnsCommand(tableId, colsToAdd)
        if sparkAdapter.isHoodieTable(tableId, sparkSession) =>
          AlterHoodieTableAddColumnsCommand(tableId, colsToAdd)
      // Rewrite the AlterTableRenameCommand to AlterHoodieTableRenameCommand
      case AlterTableRenameCommand(oldName, newName, isView)
        ......

  // 判断是否是Hudi表，通过比较provider是否等于"hudi"
  def isHoodieTable(table: CatalogTable): Boolean = {
    table.provider.map(_.toLowerCase(Locale.ROOT)).orNull == "hudi"
  }

匹配到CreateDataSourceTableCommand，并判断是Hudi表，将其转变为CreateHoodieTableCommand,这个CreateHoodieTableCommand的run方法实现了创建Hudi表的逻辑

case class CreateHoodieTableCommand(table: CatalogTable, ignoreIfExists: Boolean)
  extends HoodieLeafRunnableCommand with SparkAdapterSupport {

  override def run(sparkSession: SparkSession): Seq[Row] = {
    val tableIsExists = sparkSession.sessionState.catalog.tableExists(table.identifier)
    if (tableIsExists) {
      if (ignoreIfExists) {
        // scalastyle:off
        return Seq.empty[Row]
        // scalastyle:on
      } else {
        throw new IllegalArgumentException(s"Table ${table.identifier.unquotedString} already exists.")
      }
    }

    val hoodieCatalogTable = HoodieCatalogTable(sparkSession, table)
    // check if there are conflict between table configs defined in hoodie table and properties defined in catalog.
    CreateHoodieTableCommand.validateTblProperties(hoodieCatalogTable)

    // init hoodie table
    hoodieCatalogTable.initHoodieTable()

    try {
      // create catalog table for this hoodie table
      CreateHoodieTableCommand.createTableInCatalog(sparkSession, hoodieCatalogTable, ignoreIfExists)
    } catch {
      case NonFatal(e) =>
        logWarning(s"Failed to create catalog table in metastore: ${e.getMessage}")
    }
    Seq.empty[Row]
  }
}

optimization 和 planning

那么又是在哪里执行了run方法呢，其实是在后面的planning阶段,根据前面的学习，我们知道analysis阶段后面还有optimization和planning阶段，但是参考文章中提到

对于 SELECT 从句，sql() 方法只会触发 parsing 和 analysis 阶段。optimization和planning阶段是在show() 方法触发的

但是run方法又是在planning阶段，这就很费解，经过自己阅读源码发现，对于select语句确实是这样的，但是对于Command类型的，不用show方法也会有optimization和planning,入口为

new Dataset[Row](sparkSession, qe, RowEncoder(qe.analyzed.schema))

class Dataset[T] private[sql](
    @transient val sparkSession: SparkSession,
    @DeveloperApi @InterfaceStability.Unstable @transient val queryExecution: QueryExecution,
    encoder: Encoder[T])
  extends Serializable {

  queryExecution.assertAnalyzed()

  // Note for Spark contributors: if adding or updating any action in `Dataset`, please make sure
  // you wrap it with `withNewExecutionId` if this actions doesn't call other action.

  def this(sparkSession: SparkSession, logicalPlan: LogicalPlan, encoder: Encoder[T]) = {
    this(sparkSession, sparkSession.sessionState.executePlan(logicalPlan), encoder)
  }

  def this(sqlContext: SQLContext, logicalPlan: LogicalPlan, encoder: Encoder[T]) = {
    this(sqlContext.sparkSession, logicalPlan, encoder)
  }

  @transient private[sql] val logicalPlan: LogicalPlan = {
    // For various commands (like DDL) and queries with side effects, we force query execution
    // to happen right away to let these side effects take place eagerly.
    queryExecution.analyzed match {
      case c: Command =>
        LocalRelation(c.output, withAction("command", queryExecution)(_.executeCollect()))
      case u @ Union(children) if children.forall(_.isInstanceOf[Command]) =>
        LocalRelation(u.output, withAction("command", queryExecution)(_.executeCollect()))
      case _ =>
        queryExecution.analyzed
    }
  }

LogicalPlan

Dataset有一个变量LogicalPlan,它不是懒加载的，在new的时候就会触发，我们知道在analysis阶段返回的是CreateHoodieTableCommand,它是Command的子类，所以会匹配成功，调用

  LocalRelation(c.output, withAction("command", queryExecution)(_.executeCollect()))

  private def withAction[U](name: String, qe: QueryExecution)(action: SparkPlan => U) = {
    try {
      qe.executedPlan.foreach { plan =>
        plan.resetMetrics()
      }
      val start = System.nanoTime()
      val result = SQLExecution.withNewExecutionId(sparkSession, qe) {
        action(qe.executedPlan)
      }
      val end = System.nanoTime()
      sparkSession.listenerManager.onSuccess(name, qe, end - start)
      result
    } catch {
      case e: Exception =>
        sparkSession.listenerManager.onFailure(name, qe, e)
        throw e
    }
  }

  
  lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)

  lazy val sparkPlan: SparkPlan = {
    SparkSession.setActiveSession(sparkSession)
    // TODO: We use next(), i.e. take the first plan returned by the planner, here for now,
    //       but we will implement to choose the best plan.
    planner.plan(ReturnAnswer(optimizedPlan)).next()
  }
lazy val optimizedPlan: LogicalPlan = sparkSession.sessionState.optimizer.execute(withCachedData)   

protected def planner = sparkSession.sessionState.planner

其中的executedPlan就会先触发optimization接着触发planning，具体的逻辑也是应用各种规则/策略，参考文章已经讲的很详细了,因为这两个阶段我们没有自定义扩展规则，所以主要看一下run方法是在哪里执行的
在上面的代码中我们看到executedPlan会用到sparkPlan,继而触发planner.plan,对应planning阶段,那么需要看一下planner在哪定义的

protected def planner: SparkPlanner = {
  new SparkPlanner(session.sparkContext, conf, experimentalMethods) {
    override def extraPlanningStrategies: Seq[Strategy] =
      super.extraPlanningStrategies ++ customPlanningStrategies
  }
}

同样这里的planner是在BaseSessionStateBuilder定义的，它重写了extraPlanningStrategies,接着看一下planner.plan

planner.plan

/** A list of execution strategies that can be used by the planner */
def strategies: Seq[GenericStrategy[PhysicalPlan]]

def plan(plan: LogicalPlan): Iterator[PhysicalPlan] = {
  // Obviously a lot to do here still...

  // Collect physical plan candidates.
  val candidates = strategies.iterator.flatMap(_(plan))

  // The candidates may contain placeholders marked as [[planLater]],
  // so try to replace them by their child plans.
  val plans = candidates.flatMap { candidate =>
    val placeholders = collectPlaceholders(candidate)

    if (placeholders.isEmpty) {
      // Take the candidate as is because it does not contain placeholders.
      Iterator(candidate)
    } else {
      // Plan the logical plan marked as [[planLater]] and replace the placeholders.
      placeholders.iterator.foldLeft(Iterator(candidate)) {
        case (candidatesWithPlaceholders, (placeholder, logicalPlan)) =>
          // Plan the logical plan for the placeholder.
          val childPlans = this.plan(logicalPlan)

          candidatesWithPlaceholders.flatMap { candidateWithPlaceholders =>
            childPlans.map { childPlan =>
              // Replace the placeholder by the child plan
              candidateWithPlaceholders.transformUp {
                case p if p.eq(placeholder) => childPlan
              }
            }
          }
      }
    }
  }

  val pruned = prunePlans(plans)
  assert(pruned.hasNext, s"No plan for $plan")
  pruned
}

这个方法是将逻辑计划LogicalPlan转为物理计划PhysicalPlan,根据参考文章我们知道

上面源码的核心在于这一行：
val candidates = strategies.iterator.flatMap(_(plan))
部分同学可能看不懂这里的语法，实际上问题的答案在GenericStrategy的源码中：
策略序列的迭代器扁平化后对其中每一次都调用它的apply方法，这个方法会将逻辑计划转换成物理计划的序列。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
 * 给定一个 LogicalPlan，返回可用于执行的物理计划列表。
 * 如果此策略不适用于给定的逻辑操作，则应返回空列表
 */
abstract class GenericStrategy[PhysicalPlan <: TreeNode[PhysicalPlan]] extends Logging {

  /**
   * 返回执行计划的物理计划的占位符。
   * QueryPlanner将使用其他可用的执行策略自动填写此占位符。   
   */
  protected def planLater(plan: LogicalPlan): PhysicalPlan

  def apply(plan: LogicalPlan): Seq[PhysicalPlan]
}

也就是这里会遍历strategies并调用它的apply方法，strategies是在SparkPlanner中定义的

class SparkPlanner(val session: SparkSession, val experimentalMethods: ExperimentalMethods)
  extends SparkStrategies with SQLConfHelper {

  def numPartitions: Int = conf.numShufflePartitions

  override def strategies: Seq[Strategy] =
    experimentalMethods.extraStrategies ++
      extraPlanningStrategies ++ (
      LogicalQueryStageStrategy ::
      PythonEvals ::
      new DataSourceV2Strategy(session) ::
      FileSourceStrategy ::
      DataSourceStrategy ::
      SpecialLimits ::
      Aggregation ::
      Window ::
      JoinSelection ::
      InMemoryScans ::
      SparkScripts ::
      WithCTEStrategy ::
      BasicOperators :: Nil)

BasicOperators

其中有一个BasicOperators，它的apply方法为

object BasicOperators extends Strategy {
  def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
    case d: DataWritingCommand => DataWritingCommandExec(d, planLater(d.query)) :: Nil
    case r: RunnableCommand => ExecutedCommandExec(r) :: Nil

    case MemoryPlan(sink, output) =>
      val encoder = RowEncoder(StructType.fromAttributes(output))
      val toRow = encoder.createSerializer()
      LocalTableScanExec(output, sink.allData.map(r => toRow(r).copy())) :: Nil

我们这里的plan为CreateHoodieTableCommand是RunnableCommand的子类，所以返回ExecutedCommandExec,它是SparkPlan的子类

回到上面的withAction方法，其中会调用withNewExecutionId

def withNewExecutionId[T](
    sparkSession: SparkSession,
    queryExecution: QueryExecution)(body: => T): T = {
  val sc = sparkSession.sparkContext
  val oldExecutionId = sc.getLocalProperty(EXECUTION_ID_KEY)
  val executionId = SQLExecution.nextExecutionId
  sc.setLocalProperty(EXECUTION_ID_KEY, executionId.toString)
  executionIdToQueryExecution.put(executionId, queryExecution)
  try {
    // sparkContext.getCallSite() would first try to pick up any call site that was previously
    // set, then fall back to Utils.getCallSite(); call Utils.getCallSite() directly on
    // streaming queries would give us call site like "run at <unknown>:0"
    val callSite = sc.getCallSite()

    withSQLConfPropagated(sparkSession) {
      sc.listenerBus.post(SparkListenerSQLExecutionStart(
        executionId, callSite.shortForm, callSite.longForm, queryExecution.toString,
        SparkPlanInfo.fromSparkPlan(queryExecution.executedPlan), System.currentTimeMillis()))
      try {
        body
      } finally {
        sc.listenerBus.post(SparkListenerSQLExecutionEnd(
          executionId, System.currentTimeMillis()))
      }
    }
  } finally {
    executionIdToQueryExecution.remove(executionId)
    sc.setLocalProperty(EXECUTION_ID_KEY, oldExecutionId)
  }
}

withNewExecutionId方法会调用方法体body，这里的body为executeCollect，根据withAction方法的定义可知executeCollect为SparkPlan中的方法

executeCollect

SparkPlan.executeCollect

/**
 * Runs this query returning the result as an array.
 */
def executeCollect(): Array[InternalRow] = {
  val byteArrayRdd = getByteArrayRdd()

  val results = ArrayBuffer[InternalRow]()
  byteArrayRdd.collect().foreach { countAndBytes =>
    decodeUnsafeRows(countAndBytes._2).foreach(results.+=)
  }
  results.toArray
}

但是这里实际调用的是子类ExecutedCommandExec重写的方法

ExecutedCommandExec.executeCollect

case class ExecutedCommandExec(cmd: RunnableCommand) extends LeafExecNode {

  override lazy val metrics: Map[String, SQLMetric] = cmd.metrics

  /**
   * A concrete command should override this lazy field to wrap up any side effects caused by the
   * command or any other computation that should be evaluated exactly once. The value of this field
   * can be used as the contents of the corresponding RDD generated from the physical plan of this
   * command.
   *
   * The `execute()` method of all the physical command classes should reference `sideEffectResult`
   * so that the command can be executed eagerly right after the command query is created.
   */
  protected[sql] lazy val sideEffectResult: Seq[InternalRow] = {
    val converter = CatalystTypeConverters.createToCatalystConverter(schema)
    cmd.run(sqlContext.sparkSession).map(converter(_).asInstanceOf[InternalRow])
  }

  override def executeCollect(): Array[InternalRow] = sideEffectResult.toArray

在调用executeCollect方法时先执行sideEffectResult的逻辑，其中会执行cmd.run方法，这里的run方法是CreateHoodieTableCommand重写的run方法，实现了创建Hudi表的逻辑

普通Hive表

现在我们知道通过using hudi创建Hudi表的整个逻辑了，那么如果我们扩展了Hudi但是不是用using hudi,就想建一个普通的表，那么它的逻辑是啥，会不会因为扩展了Hudi而有问题呢？
代码中我们首先要启用Hive:.enableHiveSupport()
上面我们讲到，#createTable,#createHiveTable,他俩的区别为#createTable多了一个tableProvider,tableProvider包含关键字USING,如果sql里没有using,那么statement应该匹配为#createHiveTable,在后面的statement匹配时应该匹配到CreateHiveTableContext

public static class CreateHiveTableContext extends StatementContext {
	@Override
	public <T> T accept(ParseTreeVisitor<? extends T> visitor) {
		if ( visitor instanceof SqlBaseVisitor ) return ((SqlBaseVisitor<? extends T>)visitor).visitCreateHiveTable(this);
		else return visitor.visitChildren(this);
	}

接着调用visitCreateHiveTable,同样是在SparkSqlAstBuilder

override def visitCreateHiveTable(ctx: CreateHiveTableContext): LogicalPlan = withOrigin(ctx) {
    val (name, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)
    // TODO: implement temporary tables
    if (temp) {
      throw new ParseException(
        "CREATE TEMPORARY TABLE is not supported yet. " +
          "Please use CREATE TEMPORARY VIEW as an alternative.", ctx)
    }
    if (ctx.skewSpec.size > 0) {
      operationNotAllowed("CREATE TABLE ... SKEWED BY", ctx)
    }

   ......

    // TODO support the sql text - have a proper location for this!
    val tableDesc = CatalogTable(
      identifier = name,
      tableType = tableType,
      storage = storage,
      schema = schema,
      bucketSpec = bucketSpec,
      provider = Some(DDLUtils.HIVE_PROVIDER),
      partitionColumnNames = partitionCols.map(_.name),
      properties = properties,
      comment = Option(ctx.comment).map(string))

和visitCreateTable的逻辑差不多，其中一个区别是provider为默认为的”hive”,同样返回CreateTable

因为这里启用了Hive，我们需要看一下HiveSessionStateBuilder中的规则有啥不同

override protected def analyzer: Analyzer = new Analyzer(catalog, conf) {
  override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
    new ResolveHiveSerdeTable(session) +:
      new FindDataSourceTable(session) +:
      new ResolveSQLOnFile(session) +:
      customResolutionRules

  override val postHocResolutionRules: Seq[Rule[LogicalPlan]] =
    new DetermineTableStats(session) +:
      RelationConversions(conf, catalog) +:
      PreprocessTableCreation(session) +:
      PreprocessTableInsertion(conf) +:
      DataSourceAnalysis(conf) +:
      HiveAnalysis +:
      customPostHocResolutionRules

  override val extendedCheckRules: Seq[LogicalPlan => Unit] =
    PreWriteCheck +:
      PreReadCheck +:
      customCheckRules
}

可见多了Hive相关的规则ResolveHiveSerdeTable 和 HiveAnalysis，我们知道在上面讲的几个规则中，因为它不是datasourceTable，所以都没有匹配上，在DataSourceAnalysis后面有一个HiveAnalysis,看一下它的apply方法

object HiveAnalysis extends Rule[LogicalPlan] {
  override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
    case InsertIntoTable(r: HiveTableRelation, partSpec, query, overwrite, ifPartitionNotExists)
        if DDLUtils.isHiveTable(r.tableMeta) =>
      InsertIntoHiveTable(r.tableMeta, partSpec, query, overwrite,
        ifPartitionNotExists, query.output.map(_.name))

    case CreateTable(tableDesc, mode, None) if DDLUtils.isHiveTable(tableDesc) =>
      DDLUtils.checkDataColNames(tableDesc)
      CreateTableCommand(tableDesc, ignoreIfExists = mode == SaveMode.Ignore)

    case CreateTable(tableDesc, mode, Some(query)) if DDLUtils.isHiveTable(tableDesc) =>
      DDLUtils.checkDataColNames(tableDesc)
      CreateHiveTableAsSelectCommand(tableDesc, query, query.output.map(_.name), mode)

    case InsertIntoDir(isLocal, storage, provider, child, overwrite)
        if DDLUtils.isHiveTable(provider) =>
      val outputPath = new Path(storage.locationUri.get)
      if (overwrite) DDLUtils.verifyNotReadPath(child, outputPath)

      InsertIntoHiveDirCommand(isLocal, storage, child, overwrite, child.output.map(_.name))
  }
}

  val HIVE_PROVIDER = "hive"

  def isHiveTable(table: CatalogTable): Boolean = {
    isHiveTable(table.provider)
  }

因为他是hiveTable所以匹配到了第二个，那么转化为CreateTableCommand,在后面的customPostHocResolutionRules也就是HoodiePostAnalysisRule,也不会匹配到CreateTableCommand,也就是没有用到Hudi的逻辑，所以最终执行CreateTableCommand的run方法，也就是执行sparkSession.sessionState.catalog.createTable

case class CreateTableCommand(
    table: CatalogTable,
    ignoreIfExists: Boolean) extends RunnableCommand {

  override def run(sparkSession: SparkSession): Seq[Row] = {
    sparkSession.sessionState.catalog.createTable(table, ignoreIfExists)
    Seq.empty[Row]
  }
}

可见扩展了Hudi规则并不会影响Hive普通表的创建

总结

本文通过结合Hudi源码和Spark源码，以Create Table为例，理清了从SQL解析到最后执行的整个流程，搞懂了其中的几个关键步骤，知道了如何从Spark SQL源码跳转到Hudi Spark SQL源码并执行Hudi的建表逻辑的,并且明白了创建Hudi表和普通Hive表的区别，这样对于其他的SQL语句比如CTAS、INSERT、UPDATE、DELETE、MERGE等，也就能很容易的知道它们的逻辑了，后面的只要关注一些关键点，比如应该模式匹配到哪一类、版本之间的区别，最后再看Hudi本身的逻辑就好了

参考

本文由 董可伦 发表于伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Hudi Spark SQL源码学习总结-Create Table

本文链接：https://dongkelun.com/2022/07/20/hudiSparkSqlSourceCodeLearning/

欢迎关注我的公众号