（二开）Flink 修改源码拓展 SQL 语法

1、Flink 扩展 calcite 中的语法解析

1）定义需要的 SqlNode 节点类-以 SqlShowCatalogs 为例

a）类位置

flink/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/dql/SqlShowCatalogs.java

核心方法：

@Override
public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {writer.keyword("SHOW CATALOGS");}

b）类血缘

2）修改 includes 目录下的 .ftl 文件，在 parserImpls.ftl 文件中添加语法逻辑

a）文件位置

b）语法示例

/**
* Parse a "Show Catalogs" metadata query command.
*/
SqlShowCatalogs SqlShowCatalogs() :
{
}
{<SHOW> <CATALOGS>{return new SqlShowCatalogs(getPos());}
}

3）将 Calcite 源码中的 config.fmpp 文件复制到项目的 src/main/codegen 目录下，修改内容，来声明扩展的部分

a）文件位置

b）config.fmpp 内容

data: {# 解析器文件路径parser: tdd(../data/Parser.tdd)
}# 扩展文件的目录
freemarkerLinks: {includes: includes/
}

c）Parser.tdd 部分内容

# 生成的解析器包路径
package: "org.apache.flink.sql.parser.impl",
# 解析器名称
class: "FlinkSqlParserImpl",
# 引入的依赖类
"org.apache.flink.sql.parser.dql.SqlShowCatalogs"
# 新的关键字
keywords: ["CATALOGS"]
# 新增的语法解析方法
statementParserMethods: ["SqlShowCatalogs()"]
# 包含的扩展语法文件
implementationFiles: ["parserImpls.ftl"]

4）编译模板文件和语法文件

5）配置扩展的解析器类

withParserFactory(FlinkSqlParserImpl.FACTORY)

2、自定义扩展 Flink 的 Parser 语法

1）定义 SqlNode 类

package org.apache.flink.sql.parser.dql;import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.SqlSpecialOperator;
import org.apache.calcite.sql.SqlWriter;
import org.apache.calcite.sql.parser.SqlParserPos;import java.util.Collections;
import java.util.List;/** XSHOW CATALOGS sql call. */
public class SqlXShowCatalogs extends SqlCall {public static final SqlSpecialOperator OPERATOR =new SqlSpecialOperator("XSHOW CATALOGS", SqlKind.OTHER);public SqlXShowCatalogs(SqlParserPos pos) {super(pos);}@Overridepublic SqlOperator getOperator() {return OPERATOR;}@Overridepublic List<SqlNode> getOperandList() {return Collections.emptyList();}@Overridepublic void unparse(SqlWriter writer, int leftPrec, int rightPrec) {writer.keyword("XSHOW CATALOGS");}
}

2）修改 includes 目录下的 parserImpls.ftl 文件

/**
* Parse a "XShow Catalogs" metadata query command.
*/
SqlXShowCatalogs SqlXShowCatalogs() :
{
}
{<XSHOW> <CATALOGS>{return new SqlXShowCatalogs(getPos());}
}

3）修改 Parser.tdd 文件，新增-声明拓展的部分

imports:"org.apache.flink.sql.parser.dql.SqlXShowCatalogs"keywords:"XSHOW"statementParserMethods:"SqlXShowCatalogs()"

4）重新编译

 mvn generate-resources

5）执行测试用例

可以看到，自定义 SQL 的报错，由解析失败，变为了校验失败。

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;public class CustomFlinkSql {public static void main(String[] args) throws Exception {TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().build());// 拓展自定义语法 xshow catalogs 前// SQL parse failed. Non-query expression encountered in illegal contexttEnv.executeSql("xshow catalogs").print();// 拓展自定义语法 xshow catalogs 后// SQL validation failed. org.apache.flink.sql.parser.dql.SqlXShowCatalogs cannot be cast to org.apache.calcite.sql.SqlBasicCall}
}

6）查看生成的扩展解析器类

可以看到，在 FlinkSqlParserImpl 中，自定义的解析语法已经生成了。

3、validate 概述

在向 Flink 中添加完自定义的解析规则后，报错信息如下：

SQL validation failed. org.apache.flink.sql.parser.dql.SqlXShowCatalogs cannot be cast to org.apache.calcite.sql.SqlBasicCall

修改 validate 部分的代码

1）FlinkPlannerImpl#validate

作用：校验 SqlNode ，如果是 show catalogs 语法时直接返回。

sqlNode.isInstanceOf[SqlXShowCatalogs]

2）SqlToOperationConverter#convert

作用：将校验过的 SqlNode 转换为 Operator。

else if (validated instanceof SqlXShowCatalogs) {return Optional.of(converter.convertXShowCatalogs((SqlXShowCatalogs) validated));
}

3）SqlToOperationConverter#convertXShowCatalogs

/** Convert SHOW CATALOGS statement. */
private Operation convertXShowCatalogs(SqlXShowCatalogs sqlXShowCatalogs) {return new XShowCatalogsOperation();
}

4）XShowCatalogsOperation

package org.apache.flink.table.operations;public class XShowCatalogsOperation implements ShowOperation {@Overridepublic String asSummaryString() {return "SHOW CATALOGS";}
}

4、执行测试用例

package org.apache.flink.table.examples.java.custom;import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;public class CustomFlinkSql {public static void main(String[] args) throws Exception {TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().useBlinkPlanner().build());// FlinkSQL原本支持的语法tEnv.executeSql("show catalogs").print();// 自定义语法tEnv.executeSql("xshow catalogs").print();}
}

5、总结-FlinkSQL 的执行流程

1、对 SQL 进行校验final SqlNode validated = flinkPlanner.validate(sqlNode);2、预校验重写 Insert 语句3、调用 SqlNode.validate() 进行校验1）如果是：ExtendedSqlNode【SqlCreateHiveTable、SqlCreateTable、SqlTableLike】2）如果是：SqlKind.DDL、SqlKind.INSERT 等，无需校验，直接返回 SqlNode3）如果是：SqlRichExplain4）其它：validator.validate(sqlNode)1.校验作用域和表达式：validateScopedExpression(topNode, scope)a）将 SqlNode 进行规范化重写b）如果SQL是【TOP_LEVEL = concat(QUERY, DML, DDL)】，则在父作用域中注册查询c）校验 validateQuery i)validateFeatureii)validateNamespaceiii)validateModalityiv)validateAccessv)validateSnapshotd）如果SQL不是【TOP_LEVEL = concat(QUERY, DML, DDL)】进行类型推导2.获取校验之后的节点类型2、将 SQLNode 转换为 Operationconverter.convertSqlQuery(validated)1）生成逻辑执行计划 RelNodeRelRoot relational = planner.rel(validated);1.对查询进行转换sqlToRelConverter.convertQuery(validatedSqlNode)2）创建 PlannerQueryOperationnew PlannerQueryOperation(relational.project());3、将 Operation 转换为 List<Transformation<?>>
List<Transformation<?>> transformations = planner.translate(Collections.singletonList(modifyOperation));1）对 RelNode 逻辑执行计划进行优化，获取 optimizedRelNodesval optimizedRelNodes = optimize(relNodes)2）将 optimizedRelNodes 转换为 execGraphval execGraph = translateToExecNodeGraph(optimizedRelNodes)3）将 execGraph 转换为 transformations1.使用代码生成技术生成Function,后续可以反射调用val convertFunc = CodeGenUtils.genToInternalConverter(ctx, inputType)