Apache Flink 源码剖析 - flink-table 模块分析

graph TB
    subgraph "用户接口层"
        SQL[SQL Query]
        TABLE_API[Table API]
        CATALOG[Catalog API]
    end
    
    subgraph "解析层"
        PARSER[SQL Parser]
        VALIDATOR[SQL Validator]
        AST[Abstract Syntax Tree]
    end
    
    subgraph "优化层"
        PLANNER[Query Planner]
        OPTIMIZER[Query Optimizer]
        RULES[Optimization Rules]
    end
    
    subgraph "执行层"
        TRANSFORMATION[Transformation]
        DATASTREAM[DataStream]
    end
    
    SQL --> PARSER
    TABLE_API --> PARSER
    PARSER --> VALIDATOR
    VALIDATOR --> AST
    AST --> PLANNER
    PLANNER --> OPTIMIZER
    OPTIMIZER --> RULES
    RULES --> TRANSFORMATION
    TRANSFORMATION --> DATASTREAM

/**
 * TableEnvironment 是 Table API 和 SQL 的统一入口点
 */
@PublicEvolving
public abstract class TableEnvironment {
    
    public static final String DEFAULT_DATABASE = "default_database";
    public static final String DEFAULT_CATALOG = "default_catalog";
    
    /**
     * 创建流表环境
     */
    public static StreamTableEnvironment create(StreamExecutionEnvironment executionEnvironment) {
        return create(executionEnvironment, EnvironmentSettings.newInstance().build());
    }
    
    /**
     * 从表路径创建表
     */
    public abstract Table from(String path);
    
    /**
     * 执行 SQL 查询
     */
    public abstract Table sqlQuery(String query);
    
    /**
     * 执行 SQL 更新语句
     */
    public abstract TableResult executeSql(String statement);
    
    /**
     * 注册表
     */
    public abstract void registerTable(String name, Table table);
    
    /**
     * 注册函数
     */
    public abstract void registerFunction(String name, ScalarFunction function);
}

/**
 * StreamTableEnvironment 是流处理的表环境实现
 */
@PublicEvolving
public abstract class StreamTableEnvironment extends TableEnvironment {
    
    /**
     * 从 DataStream 创建表
     */
    public abstract <T> Table fromDataStream(DataStream<T> dataStream);
    
    /**
     * 从 DataStream 创建表并指定字段
     */
    public abstract <T> Table fromDataStream(DataStream<T> dataStream, String fields);
    
    /**
     * 将表转换为 DataStream
     */
    public abstract <T> DataStream<T> toDataStream(Table table);
    
    /**
     * 将表转换为 DataStream 并指定类型
     */
    public abstract <T> DataStream<T> toDataStream(Table table, Class<T> targetClass);
    
    /**
     * 注册 DataStream 为表
     */
    public abstract <T> void registerDataStream(String name, DataStream<T> dataStream);
    
    /**
     * 获取执行环境
     */
    public abstract StreamExecutionEnvironment getExecutionEnvironment();
}

/**
 * Table 是 Table API 的核心抽象
 */
@PublicEvolving
public interface Table {
    
    /**
     * 获取表的 Schema
     */
    TableSchema getSchema();
    
    /**
     * 选择字段
     */
    Table select(String fields);
    
    /**
     * 选择字段（表达式）
     */
    Table select(Expression... fields);
    
    /**
     * 过滤
     */
    Table where(String predicate);
    
    /**
     * 过滤（表达式）
     */
    Table where(Expression predicate);
    
    /**
     * 分组
     */
    GroupedTable groupBy(String fields);
    
    /**
     * 分组（表达式）
     */
    GroupedTable groupBy(Expression... fields);
    
    /**
     * 连接
     */
    Table join(Table right);
    
    /**
     * 连接并指定条件
     */
    Table join(Table right, String joinPredicate);
    
    /**
     * 排序
     */
    Table orderBy(String fields);
    
    /**
     * 限制
     */
    Table limit(int fetch);
    
    /**
     * 联合
     */
    Table union(Table right);
    
    /**
     * 执行并收集结果
     */
    CloseableIterator<Row> execute();
    
    /**
     * 解释执行计划
     */
    String explain();
}

/**
 * 标量函数基类
 */
@PublicEvolving
public abstract class ScalarFunction extends UserDefinedFunction {
    
    /**
     * 执行函数的方法
     * 子类必须实现一个或多个名为 eval 的方法
     */
    // public Object eval(Object... args);
    
    /**
     * 获取结果类型
     */
    @Override
    public TypeInformation<?> getResultType(Class<?>[] signature) {
        return null; // 由类型推断确定
    }
}

/**
 * 自定义标量函数示例 - 字符串长度函数
 */
public class StringLengthFunction extends ScalarFunction {
    
    /**
     * 计算字符串长度
     */
    public Integer eval(String str) {
        return str == null ? null : str.length();
    }
    
    /**
     * 支持重载 - 处理字节数组
     */
    public Integer eval(byte[] bytes) {
        return bytes == null ? null : bytes.length;
    }
}

/**
 * 复杂标量函数示例 - JSON 解析函数
 */
public class JsonExtractFunction extends ScalarFunction {
    
    private transient ObjectMapper objectMapper;
    
    @Override
    public void open(FunctionContext context) throws Exception {
        super.open(context);
        this.objectMapper = new ObjectMapper();
    }
    
    /**
     * 从 JSON 字符串中提取字段值
     */
    public String eval(String jsonStr, String path) {
        if (jsonStr == null || path == null) {
            return null;
        }
        
        try {
            JsonNode jsonNode = objectMapper.readTree(jsonStr);
            JsonNode valueNode = jsonNode.at(path);
            
            if (valueNode.isMissingNode()) {
                return null;
            }
            
            return valueNode.isTextual() ? valueNode.textValue() : valueNode.toString();
        } catch (Exception e) {
            return null;
        }
    }
}

sequenceDiagram
    participant User as 用户
    participant TableEnv as TableEnvironment
    participant Parser as SQL Parser
    participant Planner as Query Planner
    participant Optimizer as Query Optimizer
    participant Executor as Executor
    
    User->>TableEnv: sqlQuery(sql)
    TableEnv->>Parser: parse(sql)
    Parser->>Parser: 词法分析
    Parser->>Parser: 语法分析
    Parser->>TableEnv: Operation
    
    TableEnv->>Planner: translate(operation)
    Planner->>Planner: 转换为 RelNode
    Planner->>Optimizer: optimize(relNode)
    
    Optimizer->>Optimizer: 应用优化规则
    Optimizer->>Planner: 优化后的 RelNode
    
    Planner->>Executor: Transformation
    Executor->>User: Table/TableResult