Flink SQL代码补全提示（源码分析）

01 引言

Flink 源码地址： https://github.com/apache/flink

使用过Navicat的童鞋都知道，当我们写SQL的时候，工具会根据我们输入的内容弹出提示，这样可以很方便我们去写SQL，如下：

Flink也是支持SQL的，当然它也有对应的接口支持SQL提示，本文来讲讲。

02 案例 2.1 源码案例

Flink源码里面已经有代码提示的demo了，具体在CliClientTest.java（点击即可打开） ，具体在如下方法里实现了： 具体的入参为：

参数名含义statement当前输入的SQLpositionSQL末端光标的位置

返回的内容为：可能的SQL提示集合

2.2 举例

比如目前输入的SQL为：

select * fr

入参的内容为：

参数名值statement“select * fr”position11

返回的结果为集合：

FROM

接下来分析下其源码。

03 源码分析

从上面的例子，我们可以看到，获取提示的方法如下：

@Override
   public List completeStatement(String sessionId, String statement, int position) {
       receivedStatement = statement;
       receivedPosition = position;
       return Arrays.asList(helper.getSqlParser().getCompletionHints(statement, position));
   }

主要的核心是这一句：helper.getSqlParser().getCompletionHints(statement, position)。

指得是通过helper获取sql解析器，然后调用里面的getCompletionHints方法来获取提示。

接下来看看helper类。

3.1 SqlParserHelper

从下图可以得知，helper也是测试包里面的一个类： helper只是用于指引用户如何实现的一个工具类在Flink对外的API里是不存在的，这个不重要，我们看看里面的逻辑（里面已写注释）：

/**
 * SqlParser 工具类
 *
 * @author : YangLinWei
 * @createTime: 2022/9/22 2:37 下午
 * @version: 1.0.0
 */
public class SqlParserHelper {

    // 从这个TableEnvironment里获取SqlParser解析器实例
    private TableEnvironment tableEnv;

    /**
     * 构造函数
     * 

     * 使用默认EnvironmentSettings配置（当然用户可以根据自己的业务场景来设置配置）来初始化TableEnvironment
     */
    public SqlParserHelper() {
        tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
    }

    /**
     * 构造函数
     * 

     * 主要使用指定的SqlDialect来初始化TableEnvironment
     */
    public SqlParserHelper(SqlDialect sqlDialect) {
        if (sqlDialect == null || SqlDialect.DEFAULT == sqlDialect) {
            tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
        } else if (SqlDialect.HIVE == sqlDialect) {
            HiveCatalog hiveCatalog = HiveTestUtils.createHiveCatalog();
            tableEnv = TableEnvironment.create(EnvironmentSettings.newInstance().build());
            tableEnv.getConfig().setSqlDialect(sqlDialect);
            tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
            tableEnv.useCatalog(hiveCatalog.getName());
        }
    }

    /**
     * 准备一些Flink DDL 来用于测试
     */
    public void registerTables() {
        registerTable(
                "create table MyTable (a int, b bigint, c varchar(32)) "
                        + "with ('connector' = 'filesystem', 'path' = '/non', 'format' = 'csv')");
        registerTable(
                "create table MyOtherTable (a int, b bigint) "
                        + "with ('connector' = 'filesystem', 'path' = '/non', 'format' = 'csv')");
        registerTable(
                "create table MySink (a int, c varchar(32)) with ('connector' = 'COLLECTION' )");
        registerTable("create view MyView as select * from MyTable");
    }

    public void registerTable(String createTableStmt) {
        tableEnv.executeSql(createTableStmt);
    }

    /**
     * 获取SqlParser解析器
     *
     * @return Sql解析器
     */
    public Parser getSqlParser() {
        return ((TableEnvironmentInternal) tableEnv).getParser();
    }

    public TableEnvironment getTableEnv() {
        return tableEnv;
    }
}

从代码可以分析得出，如果要调用Flink里面的代码提示接口，需要先初始化TableEnviorment，然后获取Sql解析器。

接下来，看看Sql解析器的代码。

3.2 ParserImpl

从如下代码提示，可以看到解析器的实现有几个，这里使用的是blink包里面的ParserImpl。

ParserImpl 实现了Parser接口，其中Parser接口的代码及注释如下：

/**
 * SQL解析器
 * 

 * 主要解析SQL为SQL对象
 *
 * @author : YangLinWei
 * @createTime: 2022/9/22 3:00 下午
 * @version: 1.0.0
 */
@Internal
public interface Parser {

    /**
     * 解析String类型的SQL入口
     * 

     * 注意:如果创建的{@link Operation}是一个{@link QueryOperation}，
     * 它必须以{@link Planner#translate(List)}方法能够理解的形式出现。
     *
     * @param statement 待解析的SQL
     *
     * @return 将查询解析为关系 {@link Operation}s
     */
    List parse(String statement);

    /**
     * 解析SQL标识符集
     *
     * @param identifier SQL唯一标识
     *
     * @return 被解析的SQL唯一标识
     */
    UnresolvedIdentifier parseIdentifier(String identifier);

    /**
     * 解析SQL表达式入口
     *
     * @param sqlExpression 将被解析的SQL表达式
     * @param inputRowType SQL表达式中可用的字段
     * @param outputType 预期的顶级输出类型(如果可用)
     *
     * @return 解析后的表达式
     */
    ResolvedExpression parseSqlExpression(
            String sqlExpression, RowType inputRowType, @Nullable LogicalType outputType);

    /**
     * 在给定的游标位置返回给定语句的代码提示
     * 

     * 注意：补全是不区分大小写的。
     *
     * @param statement 部分或轻微错误的SQL语句
     * @param position 光标位置
     *
     * @return 当前光标位置的完成提示
     */
    String[] getCompletionHints(String statement, int position);
}

在本文，我们关注的是getCompletionHints的方法，其实现代码如下（含注释）：

/**
 * 在给定的游标位置返回给定语句的代码提示
 * 

 * 注意：补全是不区分大小写的。
 *
 * @param statement 部分或轻微错误的SQL语句
 * @param cursor 光标位置
 *
 * @return 当前光标位置的完成提示集合
 */
public String[] getCompletionHints(String statement, int cursor) {
    
    // 使用ExtendedParser来获取 代码提示的内容集
    List candidates =
            new ArrayList(
                    Arrays.asList(EXTENDED_PARSER.getCompletionHints(statement, cursor)));

    // 使用SqlAdvisor来获取 代码提示的内容集
    SqlAdvisorValidator validator = validatorSupplier.get().getSqlAdvisorValidator();
    SqlAdvisor advisor =
            new SqlAdvisor(validator, validatorSupplier.get().config().getParserConfig());
    String[] replaced = new String[1];

    List sqlHints =
            advisor.getCompletionHints(statement, cursor, replaced).stream()
                    .map(item -> item.toIdentifier().toString())
                    .collect(Collectors.toList());

    // 取代码提示的并集，并返回
    candidates.addAll(sqlHints);

    return candidates.toArray(new String[0]);
}

从代码可以得知，获取提示内容集的逻辑是分别使用“ExtendedParser”和“SqlAdvisor”来获取提示内容的集合，进一步看看这两个类。

3.2.1 ExtendedParser

ExtendedParser获取SQL提示的方法代码如下（含注释）：

public String[] getCompletionHints(String statement, int cursor) {
    // SQL 语句转为大写
    String normalizedStatement = statement.trim().toUpperCase();
    List hints = new ArrayList();
   
    // 遍历 解析策略 集合
    for (ExtendedParseStrategy strategy : PARSE_STRATEGIES) {
        // 判断查询的SQL语句是否策略里面hints集合的开头，如果是，则返回提示
        for (String hint : strategy.getHints()) {
            if (hint.startsWith(normalizedStatement) && cursor  0 && Character.isJavaIdentifierPart(sql.charAt(wordStart - 1)); --wordStart) {
        }

        if (wordStart > 0 && sql.charAt(wordStart - 1) == this.quoteStart()) {
            quoted = true;
            --wordStart;
        }

        if (wordStart