【Flink】FlinkSQL元数据验证

转x33g5p2x  于2022-02-07 转载在 Flink  
字(14.2k)|赞(0)|评价(0)|浏览(904)

1.概述

转载：FlinkSQL元数据验证

Flink1.9以后引入CatalogManager来管理Catalog和CatalogBaseTable，在执行DDL语句时将表信息封装为CatalogBaseTable存储在CatalogManager中。同时扩展了calcite的Schema接口使得Calcite在Validate阶段能够读取CatalogManager中的表信息。

2.CatalogTable写入

通过执行DDL语句，查看BlinkPlanner如何解析DDL语句，并存储到CatalogManamer中，重点查看如何解析protime as PROCTIME() 计算列的。

  1. CREATE TABLE user_address (
  2.      userId BIGINT,
  3.      addressInfo VARCHAR,
  4.      proctime AS PROCTIME()
  5. ) WITH (
  6.   'connector' = 'kafka',
  7.   'properties.bootstrap.servers' = 'localhost:9092',
  8.   'topic' = 'tp02',
  9.   'format' = 'json',
  10.   'scan.startup.mode' = 'latest-offset'
  11.  )

执行createCatalogTable调用链路。

  1. org.apache.flink.table.api.internal.TableEnvironmentImpl#executeSql
  2.     org.apache.flink.table.planner.delegation.ParserImpl#parse
  3.         org.apache.flink.table.planner.operations.SqlToOperationConverter#convert
  4.             org.apache.flink.table.planner.operations.SqlCreateTableConverter#convertCreateTable
  5.                 // 从SqlCreateTable语句中解析出CatalogTable
  6.                 org.apache.flink.table.planner.operations.SqlCreateTableConverter#createCatalogTable

从SqlCreateTable中提取TableSchema、表分区、主键、注释等信息，从而构建CatalogTableImpl。

  1. private CatalogTable createCatalogTable(SqlCreateTable sqlCreateTable) {
  2.         final TableSchema sourceTableSchema;
  3.         final List<String> sourcePartitionKeys;
  4.         final List<SqlTableLike.SqlTableLikeOption> likeOptions;
  5.         final Map<String, String> sourceProperties;
  6.         // 处理 create table like ...
  7.         if (sqlCreateTable.getTableLike().isPresent()) {
  8.             SqlTableLike sqlTableLike = sqlCreateTable.getTableLike().get();
  9.             CatalogTable table = lookupLikeSourceTable(sqlTableLike);
  10.             sourceTableSchema = table.getSchema();
  11.             sourcePartitionKeys = table.getPartitionKeys();
  12.             likeOptions = sqlTableLike.getOptions();
  13.             sourceProperties = table.getProperties();
  14.         } else {
  15.             sourceTableSchema = TableSchema.builder().build();
  16.             sourcePartitionKeys = Collections.emptyList();
  17.             likeOptions = Collections.emptyList();
  18.             sourceProperties = Collections.emptyMap();
  19.         }
  20.         //  处理SqlTableLike中的选项，INCLUDING ALL、OVERWRITING OPTIONS、EXCLUDING PARTITIONS等
  21.         Map<SqlTableLike.FeatureOption, SqlTableLike.MergingStrategy> mergingStrategies =
  22.             mergeTableLikeUtil.computeMergingStrategies(likeOptions);
  24.         Map<String, String> mergedOptions = mergeOptions(sqlCreateTable, sourceProperties, mergingStrategies);
  25.         //  提取主键
  26.         Optional<SqlTableConstraint> primaryKey = sqlCreateTable.getFullConstraints()
  27.             .stream()
  28.             .filter(SqlTableConstraint::isPrimaryKey)
  29.             .findAny();
  31.         // 获取TableSchema
  32.         TableSchema mergedSchema = mergeTableLikeUtil.mergeTables(
  33.             mergingStrategies,
  34.             sourceTableSchema,  //  非create table like 语句，sourceTableSchema为null。
  35.             sqlCreateTable.getColumnList().getList(),
  36.             sqlCreateTable.getWatermark().map(Collections::singletonList).orElseGet(Collections::emptyList),
  37.             primaryKey.orElse(null)
  38.         );
  40.         // 表分区
  41.         List<String> partitionKeys = mergePartitions(
  42.             sourcePartitionKeys,
  43.             sqlCreateTable.getPartitionKeyList(),
  44.             mergingStrategies
  45.         );
  46.         verifyPartitioningColumnsExist(mergedSchema, partitionKeys);
  47.         //  注释
  48.         String tableComment = sqlCreateTable.getComment()
  49.             .map(comment -> comment.getNlsString().getValue())
  50.             .orElse(null);
  52.         return new CatalogTableImpl(mergedSchema,
  53.             partitionKeys,
  54.             mergedOptions,
  55.             tableComment);
  56.     }

在提取TableSchema时，会将Calcite中的列的类型转换为Flink内部的数据类型。如果包含了计算列，例如procime()则会对该表达式进行验证，FlinkSqlOperatorTable类中包含了FlinkSQL的所有内置函数。

  1. /**
  2.      * Function used to access a processing time attribute.
  3.      */
  4.     public static final SqlFunction PROCTIME =
  5.         new CalciteSqlFunction(
  6.             "PROCTIME",
  7.             SqlKind.OTHER_FUNCTION,
  8.             PROCTIME_TYPE_INFERENCE,
  9.             null,
  10.             OperandTypes.NILADIC,
  11.             SqlFunctionCategory.TIMEDATE,
  12.             false
  13.         );

TableColumn生成过程：

  • 将非计算列进行类型转换，并存储到physicalFieldNamesToTypes集合。
  • 对通过表达式生成的计算列，进行存在性验证，并返回该函数对应的RelDataType。
  • 将字段的RelDataType转换为LogicalType，再转换为DataType，并构建为TableColumn。需要具体查看不同类型系统的区别。
  1. private void appendDerivedColumns(
  2.                 Map<FeatureOption, MergingStrategy> mergingStrategies,
  3.                 List<SqlNode> derivedColumns) {
  4. // 非计算列进行数据转换，存储到physicalFieldNamesToTypes
  5.             collectPhysicalFieldsTypes(derivedColumns);
  7.             for (SqlNode derivedColumn : derivedColumns) {
  8.                 final SqlTableColumn tableColumn = (SqlTableColumn) derivedColumn;
  9.                 final TableColumn column;
  10.                 if (tableColumn.isGenerated()) {
  11.                     String fieldName = tableColumn.getName().toString();
  12.                     //验证表达式，例如：proctime()函数 是否在FlinkSqlOperatorTable内注册
  13.                     SqlNode validatedExpr = sqlValidator.validateParameterizedExpression(
  14.                         tableColumn.getExpr().get(),
  15.                         physicalFieldNamesToTypes);
  16.             
  17. //  验证返回类型：proctime() 对应的RelDataType 为Flink扩展的TimeIndicatorRelDataType  
  18.                     final RelDataType validatedType = sqlValidator.getValidatedNodeType(validatedExpr);
  19.                     column = TableColumn.of(
  20.                         fieldName,
  21. //   RelDataType--->LogicalType--->DataType
  22.                         fromLogicalToDataType(toLogicalType(validatedType)),
  23.                         escapeExpressions.apply(validatedExpr));
  25.                     computedFieldNamesToTypes.put(fieldName, validatedType);
  26.                 } else {
  27.             // 非计算列转换为Flink 内部的数据类型
  28.                     String name = tableColumn.getName().getSimple();
  29. //    RelDataType --> LogicalType --> DataType
  30.                     LogicalType logicalType = FlinkTypeFactory.toLogicalType(physicalFieldNamesToTypes.get(name));
  31.                     column = TableColumn.of(name, TypeConversions.fromLogicalToDataType(logicalType));
  32.                 }
  33.                 columns.put(column.getName(), column);
  34.             }
  35.         }

计算列proctime信息，需要看下proctime Function定义时绑定的类型。

3.Validate 读取元数据

FlinkSchema包含三个子类分别为：CatalogManagerCalciteSchema，CatalogCalciteSchema，DatabaseCalciteSchema。在Calcite进行Validate时，通过调用重写的getSubSchema方法依次获取Catalog、Database信息，最终从catalogManager中获取对应的Table信息。
通过创建自定义Schema，查看Calcite获取表Schema信息。测试用例

  1. // 当前Scheam包含了USERS、JOBS两张表信息，
  2. public class CatalogManagerCalciteSchema implements Schema {
  3.     static Map<String, Table> TABLES = Maps.newHashMap();
  5.     static {
  6.         TABLES.put("USERS", new AbstractTable() { //note: add a table
  7.             @Override
  8.             public RelDataType getRowType(final RelDataTypeFactory typeFactory) {
  9.                 RelDataTypeFactory.Builder builder = typeFactory.builder();
  10.                 builder.add("ID", new BasicSqlType(new RelDataTypeSystemImpl() {
  11.                 }, SqlTypeName.INTEGER));
  12.                 builder.add("NAME", new BasicSqlType(new RelDataTypeSystemImpl() {
  13.                 }, SqlTypeName.CHAR));
  14.                 builder.add("AGE", new BasicSqlType(new RelDataTypeSystemImpl() {
  15.                 }, SqlTypeName.INTEGER));
  16.                 return builder.build();
  17.             }
  18.         });
  20.         TABLES.put("JOBS", new AbstractTable() {
  21.             @Override
  22.             public RelDataType getRowType(final RelDataTypeFactory typeFactory) {
  23.                 RelDataTypeFactory.Builder builder = typeFactory.builder();
  25.                 builder.add("ID", new BasicSqlType(new RelDataTypeSystemImpl() {
  26.                 }, SqlTypeName.INTEGER));
  27.                 builder.add("NAME", new BasicSqlType(new RelDataTypeSystemImpl() {
  28.                 }, SqlTypeName.CHAR));
  29.                 builder.add("COMPANY", new BasicSqlType(new RelDataTypeSystemImpl() {
  30.                 }, SqlTypeName.CHAR));
  31.                 return builder.build();
  32.             }
  33.         });
  34.     }
  36.     @Override
  37.     public Table getTable(String name) {
  38.         return TABLES.get(name);
  39.     }
  41.     @Override
  42.     public Set<String> getTableNames() {
  43.         return null;
  44.     }
  46.     @Override
  47.     public RelProtoDataType getType(String name) {
  48.         return null;
  49.     }
  51.     @Override
  52.     public Set<String> getTypeNames() {
  53.         return null;
  54.     }
  56.     @Override
  57.     public Collection<Function> getFunctions(String name) {
  58.         return null;
  59.     }
  61.     @Override
  62.     public Set<String> getFunctionNames() {
  63.         return Collections.emptySet();
  64.     }
  66.     @Override
  67.     public Schema getSubSchema(String name) {
  68.         return null;
  69.     }
  71.     @Override
  72.     public Set<String> getSubSchemaNames() {
  73.         return null;
  74.     }
  76.     @Override
  77.     public Expression getExpression(SchemaPlus parentSchema, String name) {
  78.         return null;
  79.     }
  81.     @Override
  82.     public boolean isMutable() {
  83.         return false;
  84.     }
  86.     @Override
  87.     public Schema snapshot(SchemaVersion version) {
  88.         return this;
  89.     }
  90. }

案例

  1. public static void main(String[] args) throws SqlParseException {
  2.         // CatalogManagerCalciteSchema是自定义的，非Flink内部的
  3.         CalciteSchema rootSchema =
  4.                 CalciteSchemaBuilder.asRootSchema(new CatalogManagerCalciteSchema());
  5.         SchemaPlus schemaPlus = rootSchema.plus();
  6.  
  7.         SqlTypeFactoryImpl factory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
  8.         //  创建CalciteCatalogReader在rel阶段时从SimpleCalciteSchema中读取元数据
  9.         CalciteCatalogReader calciteCatalogReader = new CalciteCatalogReader(
  10.                 CalciteSchema.from(schemaPlus),
  11.                 CalciteSchema.from(schemaPlus).path(null),
  12.                 factory,
  13.                 new CalciteConnectionConfigImpl(new Properties()));
  14.               
  15.         String sql = "select u.id as user_id, u.name as user_name, j.company as user_company, u.age as user_age \n"
  16.                 + "from users u join jobs j on u.name=j.name";
  18.         SqlParser parser = SqlParser.create(sql, SqlParser.Config.DEFAULT);
  19.         SqlNode sqlNode = parser.parseStmt();
  21.         SqlValidator
  22.                 validator = SqlValidatorUtil.newValidator(SqlStdOperatorTable.instance(), calciteCatalogReader, factory,
  23.                 SqlConformanceEnum.DEFAULT);
  24.         SqlNode validateSqlNode = validator.validate(sqlNode);
  25.         System.out.println(validateSqlNode);
  26.     }

Calcite通过Validate访问CatalogManagerCalciteSchema的调用链路，getSubSchema为null时代表没有子的Schema信息，则从当前Scheam读取Table信息。

Flink定义了三级Schema，通过调用getSubSchema从CatalogManager中读取Catalog、Database、Table。具体调用需要参考：org.apache.calcite.sql.validate.EmptyScope#resolve_。

CatalogManagerCalciteSchema#getSubSchemaNames：通过表名中的catalog信息，从catalogManager中获取CatalogSchema。

  1. @Override
  2.     public Schema getSubSchema(String name) {
  3.         if (catalogManager.schemaExists(name)) {
  4.             return new CatalogCalciteSchema(name, catalogManager, isStreamingMode);
  5.         } else {
  6.             return null;
  7.         }
  8.     }

CatalogCalciteSchema#getSubSchemaNames：通过表名中的database信息，从catalogManager中获取DatabaseSchema。

  1. /**
  2.      * Look up a sub-schema (database) by the given sub-schema name.
  3.      *
  4.      * @param schemaName name of sub-schema to look up
  5.      * @return the sub-schema with a given database name, or null
  6.      */
  7.     @Override
  8.     public Schema getSubSchema(String schemaName) {
  9.         if (catalogManager.schemaExists(catalogName, schemaName)) {
  10.             return new DatabaseCalciteSchema(schemaName, catalogName, catalogManager, isStreamingMode);
  11.         } else {
  12.             return null;
  13.         }
  14.     }

DatabaseSchema没有SubScheam，则从当前Schema中获取Table信息。

  1. public Table getTable(String tableName) {
  2.         ObjectIdentifier identifier = ObjectIdentifier.of(catalogName, databaseName, tableName);
  3.         return catalogManager.getTable(identifier)
  4.             .map(result -> {
  5.                 CatalogBaseTable table = result.getTable();
  6.                 FlinkStatistic statistic = getStatistic(result.isTemporary(), table, identifier);
  7.                 return new CatalogSchemaTable(
  8.                     identifier,
  9.                     result,
  10.                     statistic,
  11.                     catalogManager.getCatalog(catalogName).orElseThrow(IllegalStateException::new),
  12.                     isStreamingMode);
  13.             })
  14.             .orElse(null);
  15.     }

4.Proctime 字段验证

flinkSQL在validate读取Table schema时，会对计算列rowtime、proctime类型进行转换，转换为calcite能识别的RelDataType类型。 先列举下计算列类型转换的代码。

  1. ## CatalogManager
  2. public Optional<TableLookupResult> getTable(ObjectIdentifier objectIdentifier) {
  3.         CatalogBaseTable temporaryTable = temporaryTables.get(objectIdentifier);
  4.         if (temporaryTable != null) {
  5.             TableSchema resolvedSchema = resolveTableSchema(temporaryTable);
  6.             return Optional.of(TableLookupResult.temporary(temporaryTable, resolvedSchema));
  7.         } else {
  8.             return getPermanentTable(objectIdentifier);
  9.         }
  10.     }
  1. ## org.apache.flink.table.api.internal.CatalogTableSchemaResolver#resolve
  3.     /**
  4.      * Resolve the computed column's type for the given schema.
  5.      *
  6.      * @param tableSchema Table schema to derive table field names and data types
  7.      * @return the resolved TableSchema
  8.      */
  9.     public TableSchema resolve(TableSchema tableSchema) {
  10.         final String rowtime;
  11.         String[] fieldNames = tableSchema.getFieldNames();
  12.         DataType[] fieldTypes = tableSchema.getFieldDataTypes();
  14.         TableSchema.Builder builder = TableSchema.builder();
  15.         for (int i = 0; i < tableSchema.getFieldCount(); ++i) {
  16.             TableColumn tableColumn = tableSchema.getTableColumns().get(i);
  17.             DataType fieldType = fieldTypes[i];
  19.             if (tableColumn.isGenerated()) {
  20.         //  通过获取计算列的表达式，提取对应的DataType
  21.                 fieldType = resolveExpressionDataType(tableColumn.getExpr().get(), tableSchema);
  22.                 if (isProctime(fieldType)) {
  23.                     if (fieldNames[i].equals(rowtime)) {
  24.                         throw new TableException("Watermark can not be defined for a processing time attribute column.");
  25.                     }
  26.                 }
  27.             }
  28.                 ......
  30.             if (tableColumn.isGenerated()) {
  31.                 builder.field(fieldNames[i], fieldType, tableColumn.getExpr().get());
  32.             } else {
  33.                 builder.field(fieldNames[i], fieldType);
  34.             }
  35.         }
  36.         tableSchema.getWatermarkSpecs().forEach(builder::watermark);
  37.         tableSchema.getPrimaryKey().ifPresent(
  38.                 pk -> builder.primaryKey(pk.getName(), pk.getColumns().toArray(new String[0])));
  39.         return builder.build();
  40.     }
  1. # org.apache.flink.table.api.internal.CatalogTableSchemaResolver#resolveExpressionDataType
  2.   
  3.   private DataType resolveExpressionDataType(String expr, TableSchema tableSchema) {
  4.         ResolvedExpression resolvedExpr = parser.parseSqlExpression(expr, tableSchema);
  5.         if (resolvedExpr == null) {
  6.             throw new ValidationException("Could not resolve field expression: " + expr);
  7.         }
  8.         return resolvedExpr.getOutputDataType();
  9.     }
  10.   
  11.   # org.apache.flink.table.planner.delegation.ParserImpl#parseSqlExpression
  12.   
  13.   public ResolvedExpression parseSqlExpression(String sqlExpression, TableSchema inputSchema) {
  14.         SqlExprToRexConverter sqlExprToRexConverter = sqlExprToRexConverterCreator.apply(inputSchema);
  15.         RexNode rexNode = sqlExprToRexConverter.convertToRexNode(sqlExpression);
  16.         //    [[RelDataType]]  ---->  [[LogicalType]]
  17.         LogicalType logicalType = FlinkTypeFactory.toLogicalType(rexNode.getType());
  18.         return new RexNodeExpression(rexNode, TypeConversions.fromLogicalToDataType(logicalType));
  19.     }
  20.   
  21.   # org.apache.flink.table.planner.calcite.SqlExprToRexConverterImpl#convertToRexNodes
  22.   
  23.   public RexNode[] convertToRexNodes(String[] exprs) {
  24.         //  通过构造临时表查询，获取RexNode
  25.         String query = String.format(QUERY_FORMAT, String.join(",", exprs));
  26.         SqlNode parsed = planner.parser().parse(query);
  27.         SqlNode validated = planner.validate(parsed);
  28.     // 转换为relNode
  29.         RelNode rel = planner.rel(validated).rel;
  30.         // The plan should in the following tree
  31.         // LogicalProject
  32.         // +- TableScan
  33.         if (rel instanceof LogicalProject
  34.             && rel.getInput(0) != null
  35.             && rel.getInput(0) instanceof TableScan) {
  36.             return ((LogicalProject) rel).getProjects().toArray(new RexNode[0]);
  37.         } else {
  38.             throw new IllegalStateException("The root RelNode should be LogicalProject, but is " + rel.toString());
  39.         }
  40.     }

proctime()类型提取大致流程：

  1. 包含计算列则将建表语句中TableSchema注册为一张表temp_table。
  2. 根据建表中的计算列的表达式，例如proctime()，构建临时查询语句select proctime() from temp_table，proctime() 为Flink 内置函数。
  3. 对该查询语句进行validate，并转换RelNode，从RelNode提取行表达式RexNode。
  4. 从RexNode提取proctime() 对应的RelDataType，最终转换为DataType。

[FLINK-18378]之前对计算列的处理流程。根据proctime、rowtime单独做了区分。
疑问：在DDL语句中已经将proctime转换为DataType，在validate获取Table schema是直接拿fieldType即可，为什么还要做一次解析。

  1. for (int i = 0; i < tableSchema.getFieldCount(); ++i) {
  2.             TableColumn tableColumn = tableSchema.getTableColumns().get(i);
  3.             DataType fieldType = fieldTypes[i];
  4.             if (tableColumn.isGenerated() && isProctimeType(tableColumn.getExpr().get(), tableSchema)) {
  5.                 if (fieldNames[i].equals(rowtime)) {
  6.                     throw new TableException("Watermark can not be defined for a processing time attribute column.");
  7.                 }
  8.                 TimestampType originalType = (TimestampType) fieldType.getLogicalType();
  9.                 LogicalType proctimeType = new TimestampType(
  10.                         originalType.isNullable(),
  11.                         TimestampKind.PROCTIME,
  12.                         originalType.getPrecision());
  13.                 fieldType = TypeConversions.fromLogicalToDataType(proctimeType);
  14.             } else if (isStreamingMode && fieldNames[i].equals(rowtime)) {
  15.                 TimestampType originalType = (TimestampType) fieldType.getLogicalType();
  16.                 LogicalType rowtimeType = new TimestampType(
  17.                         originalType.isNullable(),
  18.                         TimestampKind.ROWTIME,
  19.                         originalType.getPrecision());
  20.                 fieldType = TypeConversions.fromLogicalToDataType(rowtimeType);
  21.             }
  22.             if (tableColumn.isGenerated()) {
  23.                 builder.field(fieldNames[i], fieldType, tableColumn.getExpr().get());
  24.             } else {
  25.                 builder.field(fieldNames[i], fieldType);
  26.             }
  27.         }

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/qq_21383435/article/details/122769008

内容来源于网络，如有侵权，请联系作者删除！

flinkkafkaapache

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com