本文主要描述 Flink1.9 新提供的 create table sql 通过 Calcite 解析，校验 并注册到catalog的过程。
样例SQL:
CREATE TABLE user_log (
user_id VARCHAR,
item_id VARCHAR,
category_id VARCHAR,
behavior VARCHAR,
ts TIMESTAMP
) WITH (
'connector.type' = 'kafka', -- 使用 kafka connector
'connector.version' = 'universal', -- kafka 版本，universal 支持 0.11 以上的版本
'connector.topic' = 'xc_user_behavior', -- kafka topic
'connector.startup-mode' = 'earliest-offset', -- 从起始 offset 开始读取
'connector.properties.0.key' = 'zookeeper.connect', -- 连接信息
'connector.properties.0.value' = '172.16.8.107:2181',
'connector.properties.1.key' = 'bootstrap.servers',
'connector.properties.1.value' = '172.16.8.107:9092',
'update-mode' = 'append',
'format.type' = 'json', -- 数据源格式为 json
'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则
)
带着两个疑问：1:如何自定义create table 语法(之后会另外写一篇)， 2：create table 的表信息是如何注册到catalog上的，供之后的select 语句查询；
一：StreamTableEnvironment.create() 关注：Planner 的初始化
-
实例化一个CatalogManager（元数据管理器）
--初始化一个GenericInMemroyCatalog作为默认的Catalog
- 实例化StreamPlanner(本文以流作为背景，所以实例化目标为StreamPlanner)
- 初始化StreamPlanner继承的超类PlannerContext
- 可以看到提供了统一生成Calcite FrameworkConfig的函数createFrameworkConfig，关注其中对defaultSchema的设置，（之前实例的CatalogManager的包装类CatalogManagerCalciteSchema作为默认的Schema)，另外ParserFactory参数也需要关注下，FlinkSqlParserImpl.FACTORY 这个是flink 基于Calcite 扩展的支持DDL的解析类；
-
初始化查询优化器:
RelOptPlanner planner = new VolcanoPlanner(frameworkConfig.getCostFactory(), frameworkConfig.getContext());//基于cost的优化器
到这一步Planner环境初始化相关的操作已经完成。
二：解析Create table Sql语句:StreamTableEnvironment.sqlUpdate(createSqlStr)
-
StreamTableEnvironment 调用 之前初始化环境过程中创建的 StreamPlanner 解析create table 语句 生成 CreateTableOperation;
其中生成Operation 的部分主体是：
- PlannerContext基于统一提供的createFrameworkConfig() 生成FlinkPlannerImpl。
- FlinkPlannerImpl 解析 DDL语句 生成SqlNode（如何根据Calcite自定义Table create 这个会后会另外单独写一篇）。
- 调用validate() 对create table 信息进行校验， 包括table 的column, primaryKeys,uniqueKeys，partitionKey设置是否正确,这个校验不包括对create table 语法的校验，create 语法的校验在parse阶段已经处理过；
- org.apache.flink.table.planner.operations.SqlToOperationConverter#convert：将SqlNode转换为org.apache.flink.table.operations.ddl.CreateTableOperation，当然convert()方法包含对Query, Create, Drop, Insert 的处理，本文只关注convertCreateTable模块；
- 调用TableEnvironmentImpl#registerCatalogTableInternal 把生成的CreateTableOperation注册到CatalogManager 上, 注册的元数据信息之后可以被sql 查询引用；
调用GenericInMemoryCatalog#createTable 将表的信息添加到catalog 元数据上；
