Flink kernel source code (VII) Flink SQL submission process

Let's learn the seventh chapter Flink SQL Resolution submission process of .

Problem sorting ：

1. Flink Medium Calcite What is it? ？
2. Flink SQL What is the submission process of ？

1. Calcite

Apache Calcite It's a dynamic data management framework , It has many typical database management system functions , Such as SQL analysis 、SQL check 、SQL Query optimization, etc , Some functions are omitted , If relevant data is not stored , It also does not completely include relevant processing data, etc .

Flink Medium sql analysis 、sql The checksum sql Optimization is dependency calcite To complete .

Sort out Calcite SQL Several stages of implementation ：

1. adopt Parser The parser will pass in sql Parse into a lexical tree ,SqlNode As a node of the tree
2. Do lexical verification Validate, Type verification , Metadata verification and so on
3. The calibrated SqlNode The tree is transformed into the corresponding relational algebraic expression , It's also a tree ,RelNode As node
4. take RelNode Relational algebra expression tree , Through two built-in optimizers Volcano , Hep By optimizing the expression of relational algebra, a tree of optimal logic algebra is obtained , It's also RelNode
5. Optimal logical algebraic expression （RelNode）, Will be converted into the corresponding executable physical execution plan （ The transformation logic varies according to the framework ）, image Flink Just turn into his Operator To run

2. Flink SQL Submission process

First overall Flink SQL Submit a description of the process , Then explain in detail from the perspective of source code .

There are two stages in total ：

1. sql To operation Transformation

• SQL analysis ： call parser Method , take SQL Turn to unverified AST Abstract syntax tree , That is to say SqlNode, It mainly uses lexical parsing and grammatical parsing . Lexical analysis is to Sql Statement into a group token, Grammatical parsing is about token Perform recursive descent lexical analysis

• SQL check ： It is to verify the unchecked abstract syntax tree into a verified abstract syntax tree , In the verification stage, two parts are mainly verified ：1） Checklist name , Field name , Whether the function name is correct 2） Verify that the particular type is correct , Such as join operation ,groupby Whether there is nesting, etc

• call rel() Method ： The abstract syntax tree SqlNode Into a relational algebra tree RelNode（ Relational algebra expression ） and RexNode Row expression , In the process ,DDL It is not implemented rel() Methodical , because DDL It is actually a modification of the element area , No complex queries involved

• call convert() Method ： In the end RelNode Turn into operation,operation It includes many types , But eventually the root node will be generated modify operation

2. operation To transformations Transformation

• take modify operation Finally, it's converted into calcite Logical plan tree （calcite logicalPlan）, secondly , take calcite logicalPlan To flink Logical plan tree （Flink LogicalRel）
• call optimize() Method , take Flink LogicalRel Optimize into physical plan FlinkPhysicalRel, Including two optimization rules ： Rule based optimization RBO And cost based optimization CBO
• call translateToExecNodeGraph Method , The method is to convert the physical plan into ExecGraph
• call translateToPlan() Method , Will eventually ExecGraph Turn into transformations

3. The source code parsing

3.1 Sql Statement parsing idiom tree stage （SQL - > SqlNode）

TableEnvironmentImpl yes sql Entry class for execution ,TableEnvironmentImpl Provided in executeSql,sqlQuery And other methods are used to perform DDL and DML etc. sql,sql The execution will be on sql To analyze ,ParserImpl yes flink call sql Parsed implementation class ,ParserImpl#parse() Method by calling the wrapper object CalciteParser#parse() Method and create and call using javacc Generated sql Parser (FlinkSqlParserImpl) Medium parseSqlStmtEof Method to complete sql analysis , And back to SqlNode object

Core code ：

public List<Operation> parse(String statement) {
    
	CalciteParser parser = calciteParserSupplier.get();
	FlinkPlannerImpl planner = validatorSupplier.get();
	//TODO  Call here to use javacc Generated parser , take sql Statement parsing into sqlNode
	SqlNode parsed = parser.parse(statement);
 
	//TODO  take sqlNode Convert to Operation
	Operation operation = SqlToOperationConverter.convert(planner, catalogManager, parsed)
		.orElseThrow(() -> new TableException("Unsupported query: " + statement));
	return Collections.singletonList(operation);
}

among parser.parse(…) Method , take sql Statement parsing into sqlNode. Corresponding table name 、 Name 、with Property parameters 、 Primary key 、 The only key 、 The partitioning key 、 watermark 、 Table annotation 、 Table operations （create table、alter table、drop table） in SqlNode Object's corresponding properties ,SqlNode It's a tree structure, that is AST.

3.2 Sql check （SqlNode - > Operation）

sql Execute after parsing sql check ,flink sql Added in SqlNode Convert to Operation The process of ,sql The verification is completed in this process . stay SqlToOperationConverter#convert() Method to complete the conversion of this process , Will pass through FlinkPlannerInpm#validate() Method to set the table 、 function 、 Fields, etc. are verified and based on the generated validated SqlNode Generate corresponding Opeation.

Different sql after convert After processing, return different Operation, Finally, according to different Operation There are different handling behaviors .

3.3 Flink SQL Optimize （Operation - > RelNode->Transformation ）

Blink There is no direct use of Calcite The optimizer for , But through the combination of rules and Calcite The optimized combination is batch and stream Implements a custom optimizer .
Before the optimization is implemented, the SqlNode To RelNode, be based on RelNode call PlannerBase#optimize() And implement StreamCommonSubGraphBasedOptimizer#doOptimize() Method to complete the optimization

At the completion of Sql To RelNode After the conversion of , Will execute executeOperation(…) operation , Here we will first sqlNode convert to RelNode. Then optimize the operation .

And then according to the incoming sql Statement type , Different operation options . contains Modify、CreateTable、DropTable etc. .

ad locum , There are conversion and optimization operations , The focus is on translate In the method , The final call is PlannerBase Inside translate(...) Method

override def translate(
    modifyOperations: util.List[ModifyOperation]): util.List[Transformation[_]] = {
    
  if (modifyOperations.isEmpty) {
    
    return List.empty[Transformation[_]]
  }
  // prepare the execEnv before translating
  getExecEnv.configure(
    getTableConfig.getConfiguration,
    Thread.currentThread().getContextClassLoader)
  overrideEnvParallelism()
 
  // TODO  Complete the conversion here  SqlNode Convert to RelNode
  val relNodes = modifyOperations.map(translateToRel)
 
  // TODO  Complete the optimization here 
  val optimizedRelNodes = optimize(relNodes)
  val execNodes = translateToExecNodePlan(optimizedRelNodes)
  translateToPlan(execNodes)
}

In the end by the translateToPlan Methods will ExecNode convert to Transfomation list

Based on generated Transformation Object call StreamExecutor#createPipeline() Method generation StreamGraph Then you can perform the task .

Reference resources ：

• FlinkSQL The source code parsing （ One ） Conversion process
• Silicon Valley Flink Kernel source code analysis course （ From entry to mastery ）