1、SqlParser解析阶段(SQL -> SqlNode)

2、Validator验证阶段

3、逻辑计划(SqlNode -> RelNode/RexNode)

4、优化器(RelNode -> LogicalNode -> ExecNode)

那么如何解决呢？首先想到的思路就是将这段共用逻辑注册成表，这样Flink就能知道这段逻辑是共用的。

目前有“注册视图”（create view as query）和“注册表”（registerTable）两种方式能够将共用逻辑注册成表。在Flink中，当执行‘create view as query' 创建视图或者调用registerTable注册表时，底层都会在catalog中创建临时表，区别在于create view创建表的实现类为CatalogViewImpl，而registerTable创建表的实现类为QueryOperationCatalogView。

前者CatalogViewImpl的查询逻辑使用字符串表示，而后者 QueryOperationCatalogView的查询逻辑已经被解析为QueryOperation。也就是说，执行创建视图的语句时，最终创建的临时表仅仅是缓存了查询部分的SQL语句，当其他命令使用这个临时表时还需要重新解析临时表中的查询语句，而重新解析带来的问题就是创建新的RelNode，产生不同的摘要，这样Flink仍然不能够识别到这段共用逻辑并复用。

相反，regiterTable这样的方式就不需要对临时表中的查询语句进行重新解析。因此可以采用regiterTable将共用逻辑注册成表。示例代码如下：



但是，为了SQL化，就需要有语法去支持中间表注册，以屏蔽底层的api调用，实现用户无感知。新语法预期如下：


如何实现新的语法，来支持中间表注册呢？目前有2种解决方案：

方案1：框架先使用正则匹配判断SQL类型，之后提取出临时表名和查询逻辑，比如上面的SQL经过正则匹配提取组之后可以得到表名为tmp，查询逻辑为'select SBSTR(data, 0, 6) data,topic from source_table'，之后框架去调用table api进行注册。

方案2：修改flink-table模块源码扩展语法，实现对register table语法的支持。

从实现难度上来说，方案1的改动少，难度也较小，而方案2虽然改动较大，但是通用性更好。下面主要围绕方案2的实现展开。

register table语法扩展大致分为以下3个步骤：

即修改Java CC相关文件，使得 SqlParser可以识别新的语法并解析为AST。

✦ 1.1 增加关键字“REGISTER”

首先需要让解析器识别新的关键字“REGISTER”，因此修改Parser.tdd，在keywords和nonReservedKeywords中分别增加“REGISTER”关键字。

✦ 1.2 创建SqlNode (SqlRegisterTable)

由于SqlParser解析SQL生成的AST数据类型为SqlNode，因此需要增加相应的SqlNode。在org.apache.flink.sql.parser.ddl下创建SqlRegisterTable：

package org.apache.flink.sql.parser.ddl;
import org.apache.calcite.sql.*,

import org.apache.calcite.sql.parser.SqlParserPos;

import javax.annotation.Nonnull;

import java.util.Collections;import java.util.List;
public class SqlRegisterTable extends SqlCall {

public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("REGISTER TABLE", SqlKind.OTHER_DDL);
private SqlIdentifier tableName;

private SqlNode query;    

 // 部分代码省略

 }

修改Parser.tdd，在imports增加org.apache.flink.sql.parser.ddl.SqlRegisterTable，即刚才新建的 SqlRegisterTable的全类名路径。

✦ 1.3 增加语法解析模版

除了关键字和SqlNode，还需要相应的语法模版，让解析器能够把SQL解析为SqlRegisterTable。

修改parserImpls.ftl，增加语法解析模版：

private Operation convertRegisterTable(SqlRegisterTable sqlRegisterTable) { 
   UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(sqlRegisterTable.fullTableName()); 
   ObjectIdentifier identifier = catalogManager.qualifyIdentifier(unresolvedIdentifier);
    PlannerQueryOperation operation = toQueryOperation(flinkPlanner, validateQuery);
    return new RegisterTableOperation(identifier, operation); 
 }