Flink源码

2021-11-22 09:24:04 浏览数 (1)

下面我们以一个具体例子来说怎么做SQL语法拓展，比如我们需要支持语法

代码语言：txt复制
USE FUNCTION func_name [ WITH (name = '', version = '') ]

1. 创建maven项目

2. 复制模板文件

从Flink源码flink-table/flink-sql-parser/src/main/codegen文件夹copy到自己项目的src/main目录下

3. 拓展SQL语法

主要分为下面三步

3.1 在 codegen/includes/parserImpls.ftl 定义解析use function 的规则，包含了方法名及对应的规则。

代码语言：txt复制
/**
代码语言：txt复制
*  USE FUNCTION func_name [ WITH (name = '', version = '') ]
*/
代码语言：txt复制
SqlCall SqlUseFunction() :
代码语言：txt复制
{
代码语言：txt复制
     SqlParserPos pos;
代码语言：txt复制
     SqlIdentifier funcName;
代码语言：txt复制
     SqlNodeList funcProps = null;
代码语言：txt复制
}
代码语言：txt复制
{
代码语言：txt复制
     <USE>
代码语言：txt复制
     {
代码语言：txt复制
         // Token位置
代码语言：txt复制
         pos = getPos();
代码语言：txt复制
     }
代码语言：txt复制
     <FUNCTION>
代码语言：txt复制
         funcName = CompoundIdentifier()
代码语言：txt复制
     [ <WITH> funcProps = TableProperties() ]
代码语言：txt复制
     {
代码语言：txt复制
         return new SqlUseFunction(pos, funcName, funcProps);
代码语言：txt复制
     }
代码语言：txt复制
}

3.2 拓展 SqlCall

规则匹配成功返回一个SqlUseFunction节点，作为解析树中的SqlNode。

代码语言：txt复制
public class SqlUseFunction extends SqlCall {
代码语言：txt复制
    private static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("USE FUNCTION",
代码语言：txt复制
            SqlKind.OTHER_FUNCTION);
代码语言：txt复制
    private final SqlIdentifier funcName;
代码语言：txt复制
    private final SqlNodeList funcProps;
代码语言：txt复制
    /**
代码语言：txt复制
     * SqlUseFunction constructor.
     *
     * @param pos sql define location
     * @param funcName function name
     * @param funcProps function property
     * */
    public SqlUseFunction(SqlParserPos pos, SqlIdentifier funcName, SqlNodeList funcProps) {
        super(pos);
        this.funcName = funcName;
        this.funcProps = funcProps;
    }
代码语言：txt复制
    @Override
代码语言：txt复制
    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
代码语言：txt复制
        writer.keyword("USE FUNCTION");
代码语言：txt复制
        funcName.unparse(writer, leftPrec, rightPrec);
代码语言：txt复制
        if (funcProps != null) {
代码语言：txt复制
            writer.keyword("WITH");
代码语言：txt复制
            SqlWriter.Frame frame = writer.startList("(", ")");
代码语言：txt复制
            for (SqlNode c : funcProps) {
代码语言：txt复制
                writer.sep(",");
代码语言：txt复制
                c.unparse(writer, 0, 0);
代码语言：txt复制
            }
代码语言：txt复制
            writer.endList(frame);
代码语言：txt复制
        }
代码语言：txt复制
    }
代码语言：txt复制
    @Override
代码语言：txt复制
    public SqlOperator getOperator() {
代码语言：txt复制
        return OPERATOR;
代码语言：txt复制
    }
代码语言：txt复制
    @Override
代码语言：txt复制
    public List<SqlNode> getOperandList() {
代码语言：txt复制
        return ImmutableNullableList.of(funcName, funcProps);
代码语言：txt复制
    }
代码语言：txt复制
}

3.3 将添加的规则，加入配置文件 codegen/data/Parser.tdd

  • imports增加SqlUseFunction类。
  • statementParserMethods增加定义的规则方法。在Parser中使用新规则。
代码语言：txt复制
 imports: [
代码语言：txt复制
    "org.apache.flink.sql.extended.SqlUseFunction"
代码语言：txt复制
 ]
代码语言：txt复制
  # List of methods for parsing custom SQL statements.
代码语言：txt复制
  # Return type of method implementation should be 'SqlNode'.
代码语言：txt复制
  # Example: SqlShowDatabases(), SqlShowTables().
代码语言：txt复制
  statementParserMethods: [
代码语言：txt复制
    "SqlUseFunction()"
代码语言：txt复制
  ]
  1. 编译undefined执行mvn clean compile生成定义的解析类SqlParserImpl.java。重点看生成的SqlUseFunction()，编译成功就可以在sql中使用该语法了。
flink sql 大数据 打包

0 人点赞