Flink SQL UDF

1 函数

1.1 函数的定义

特性​​	​BuiltInFunctionDefinition​​	​FlinkSqlOperatorTable​​
​​核心功能​​	函数元数据管理、类型推导、执行逻辑封装	SQL语法解析、函数符号表管理
​​适用阶段​​	执行阶段（运行时）	解析阶段（编译时）
​​与Calcite的关系​​	无直接依赖	继承自Calcite的SqlOperatorTable
​​扩展性​​	支持用户自定义函数（UDF）	主要管理内置运算符和标准SQL函数

1.1.1 BuiltInFunctionDefinition​

属于Table API/SQL模块的​​函数定义层​​，是Flink对函数的统一抽象表示。
它定义了函数的基本属性（如名称、类型、参数策略等），与具体的执行引擎解耦，适用于所有API层（包括DataStream API和Table API）。

​​作用​​：

• 提供函数的元信息（如输入/输出类型策略、确定性等）
• 支持跨API的统一函数注册（如通过TableEnvironment.registerFunction()）
• 作为SQL和Table API函数调用的中间表示

1.1.2 FlinkSqlOperatorTable​

Operator table that contains only Flink-specific functions and operators.

是​​SQL解析层​​的组件，继承自Apache Calcite的SqlOperatorTable，专门用于SQL语法解析和验证阶段。
它定义了SQL标准中的运算符和函数（如+, SUM, OVER等）如何映射到Flink的执行逻辑。

​​作用​​：

• 提供SQL语法树（AST）中运算符和函数的符号表
• 在SQL解析阶段验证函数是否存在及参数合法性
• 将SQL函数名映射到BuiltInFunctionDefinition或其他执行逻辑

1.1.3 协同工作流程​

1. ​​SQL解析阶段​​：当用户提交SQL查询时，FlinkSqlOperatorTable负责识别函数名（如COUNT）并验证语法合法性
2. ​​逻辑计划生成​​：解析通过后，SQL函数会被转换为BuiltInFunctionDefinition实例，作为逻辑计划的一部分
3. ​​执行阶段​​：BuiltInFunctionDefinition提供具体的类型推导和执行逻辑，最终生成Flink作业图（JobGraph）

1.2 函数的校验

以IP_TO_LONG函数为例：

public static final SqlFunction IP_TO_LONG =  
        new SqlFunction(  
                "IpToLong",  // name
                SqlKind.OTHER_FUNCTION,  // SqlKind
                ReturnTypes.BIGINT_NULLABLE,  // SqlReturnTypeInference
                null,  // SqlOperandTypeInference
                OperandTypes.STRING,  // SqlOperandTypeChecker
                SqlFunctionCategory.STRING // SqlFunctionCategory
                );

// org.apache.calcite.sql.SqlOperator#checkOperandTypes
// org.apache.calcite.sql.type.FamilyOperandTypeChecker#checkOperandTypes
// org.apache.calcite.sql.type.FamilyOperandTypeChecker#checkSingleOperandType

...
family.getTypeNames().contains(typeName)
...

1.3 查找所有的函数定义

List<SqlOperator> operatorList = FlinkSqlOperatorTable.instance().getOperatorList();  
  
for (int i = 0; i < operatorList.size(); i++) {  
    SqlOperator sqlOperator = operatorList.get(i);  
    System.out.printf("FlinkSqlOperateTable: %03d,\t%-35s,%6s,\n",  
            i + 1,  
            sqlOperator.getName(),  
            sqlOperator.kind);  
}