本文将简述Flink SQL / Table API的内部实现，为大家把 “从SQL语句到具体执行” 这个流程串起来。并且尽量多提供调用栈，这样大家在遇到问题时就知道应该从什么地方设置断点，对整体架构理解也能更加深入。

目录

本文将简述Flink SQL / Table API的内部实现，为大家把 “从SQL语句到具体执行” 这个流程串起来。并且尽量多提供调用栈，这样大家在遇到问题时就知道应该从什么地方设置断点，对整体架构理解也能更加深入。

SQL流程中涉及到几个重要的节点举例如下：

  1. // NOTE : 执行顺序是从上至下, " -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |      
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |     
  12. *     SqlToRelConverter.convertQuery // 语义分析，生成逻辑计划，作用是SqlNode–>RelNode
  13. *        |    
  14. *        |     
  15. *        +-----> LogicalProject (RelNode) // Abstract Syntax Tree，未优化的RelNode   
  16. *        |      
  17. *        |     
  18. *    FlinkLogicalJoinConverter (RelOptRule) // Flink定制的优化rules      
  19. *    VolcanoRuleCall.onMatch // 基于Flink定制的一些优化rules去优化 Logical Plan 
  20. *        | 
  21. *        |   
  22. *        +-----> FlinkLogicalJoin (RelNode)  // Optimized Logical Plan，逻辑执行计划
  23. *        |  
  24. *        |    
  25. *    StreamExecJoinRule (RelOptRule) // Rule that converts FlinkLogicalJoin without window bounds in join condition to StreamExecJoin     
  26. *    VolcanoRuleCall.onMatch // 基于Flink rules将optimized LogicalPlan转成Flink物理执行计划
  27. *        |       
  28. *        |   
  29. *        +-----> StreamExecJoin (FlinkRelNode) // Stream physical RelNode，物理执行计划
  30. *        |      
  31. *        |     
  32. *    StreamExecJoin.translateToPlanInternal  // 作用是生成 StreamOperator, 即Flink算子  
  33. *        |     
  34. *        |     
  35. *        +-----> StreamingJoinOperator (StreamOperator) // Streaming unbounded Join operator in StreamTask   
  36. *        |     
  37. *        |       
  38. *    StreamTwoInputProcessor.processRecord1// 在TwoInputStreamTask调用StreamingJoinOperator，真实的执行 
  39. *        |
  40. *        |

后续我们会以这个图为脉络进行讲解。

Flink Table API&SQL 为流式数据和静态数据的关系查询保留统一的接口，而且利用了Apache Calcite的查询优化框架和SQL parser。

为什么Flink要使用Table API呢？总结来说，关系型API的好处如下：

  • 关系型API是声明式的
  • 查询能够被有效的优化
  • 查询可以高效的执行
  • “Everybody” knows SQL

Calcite是这里面的核心成员。Apache Calcite是面向Hadoop新的sql引擎，它提供了标准的SQL语言、多种查询优化和连接各种数据源的能力。

下面是 Calcite 概念梳理：

  • 关系代数（Relational algebra）：即关系表达式。它们通常以动词命名，例如 Sort, Join, Project, Filter, Scan, Sample.
  • 表达式有各种特征（Trait）：使用 Trait 的 satisfies() 方法来测试某个表达式是否符合某 Trait 或 Convention.
  • 规则（Rules）：用于将一个表达式转换（Transform）为另一个表达式。它有一个由 RelOptRuleOperand 组成的列表来决定是否可将规则应用于树的某部分。
  • 规划器（Planner） ：即请求优化器，它可以根据一系列规则和成本模型（例如基于成本的优化模型 VolcanoPlanner、启发式优化模型 HepPlanner）来将一个表达式转为语义等价（但效率更优）的另一个表达式。
  • RelNode ：代表了对数据的一个处理操作，常见的操作有 Sort、Join、Project、Filter、Scan 等。它蕴含的是对整个 Relation 的操作，而不是对具体数据的处理逻辑。RelNode 会标识其 input RelNode 信息，这样就构成了一棵 RelNode 树。
  • RexNode : 行表达式（标量表达式），蕴含的是对一行数据的处理逻辑。每个行表达式都有数据的类型。这是因为在 Valdiation 的过程中，编译器会推导出表达式的结果类型。常见的行表达式包括字面量 RexLiteral， 变量 RexVariable，函数或操作符调用 RexCall 等。RexNode 通过 RexBuilder 进行构建。
  • RelTrait : 用来定义逻辑表的物理相关属性（physical property），三种主要的 trait 类型是：Convention、RelCollation、RelDistribution；

Sql 的执行过程一般可以分为四个阶段，Calcite 与这个很类似，但Calcite是分成五个阶段 ：

  1. SQL 解析阶段，生成AST（抽象语法树）（SQL–>SqlNode）

  2. SqlNode 验证（SqlNode–>SqlNode）

  3. 语义分析，生成逻辑计划（Logical Plan）（SqlNode–>RelNode/RexNode）

  4. 优化阶段，按照相应的规则（Rule）进行优化（RelNode–>RelNode）

  5. 生成ExecutionPlan，生成物理执行计划（DataStream Plan）

Flink承载了 Table API 和 SQL API 两套表达方式。它以Apache Calcite这个SQL解析器做SQL语义解析，统一生成为 Calcite Logical Plan（SqlNode 树）；随后验证；再利用 Calcite的优化器优化转换规则和logical plan，根据数据源的性质(流和批)使用不同的规则进行优化，优化为 RelNode 逻辑执行计划树；最终优化后的plan转成常规的Flink DataSet 或 DataStream 程序。任何对于DataStream API和DataSet API的性能调优提升都能够自动地提升Table API或者SQL查询的效率。

一条stream sql从提交到calcite解析、优化最后到Flink引擎执行，一般分为以下几个阶段:

  1. Sql Parser: 将sql语句通过java cc解析成AST(语法树),在calcite中用SqlNode表示AST;
  2. Sql Validator: 结合数字字典(catalog)去验证sql语法；
  3. 生成Logical Plan: 将sqlNode表示的AST转换成LogicalPlan, 用relNode表示;
  4. 生成 optimized LogicalPlan: 先基于calcite rules 去优化logical Plan, 再基于Flink定制的一些优化rules去优化logical Plan；
  5. 生成Flink PhysicalPlan: 这里也是基于Flink里头的rules，将optimized LogicalPlan转成成Flink的物理执行计划；
  6. 将物理执行计划转成Flink ExecutionPlan: 就是调用相应的tanslateToPlan方法转换和利用CodeGen元编程成Flink的各种算子。

而如果是通过table api来提交任务的话，也会经过calcite优化等阶段，基本流程和直接运行sql类似:

  1. table api parser: Flink会把table api表达的计算逻辑也表示成一颗树，用treeNode去表式;
    在这棵树上的每个节点的计算逻辑用Expression来表示。
  2. Validate: 会结合数字字典(catalog)将树的每个节点的Unresolved Expression进行绑定，生成Resolved Expression；
  3. 生成Logical Plan: 依次遍历数的每个节点，调用construct方法将原先用treeNode表达的节点转成成用calcite 内部的数据结构relNode 来表达。即生成了LogicalPlan, 用relNode表示;
  4. 生成 optimized LogicalPlan: 先基于calcite rules 去优化logical Plan,
    再基于Flink定制的一些优化rules去优化logical Plan；
  5. 生成Flink PhysicalPlan: 这里也是基于Flink里头的rules，将optimized LogicalPlan转成成Flink的物理执行计划；
  6. 将物理执行计划转成Flink ExecutionPlan: 就是调用相应的tanslateToPlan方法转换和利用CodeGen元编程成Flink的各种算子。

可以看出来，Table API 与 SQL 在获取 RelNode 之后是一样的流程，只是获取 RelNode 的方式有所区别：

  • Table API ：通过使用 RelBuilder来拿到RelNode（LogicalNode与Expression分别转换成RelNode与RexNode）；
  • SQL ：通过使用Planner。首先通过parse方法将用户使用的SQL文本转换成由SqlNode表示的parse tree。接着通过validate方法，使用元信息来resolve字段，确定类型，验证有效性等等。最后通过rel方法将SqlNode转换成RelNode；

TableEnvironment对象是Table API和SQL集成的一个核心，支持以下场景：

  • 注册一个Table。
  • 将一个TableSource注册给TableEnvironment，这里的TableSource指的是将数据存储系统的作为Table，例如mysql,hbase,CSV,Kakfa,RabbitMQ等等。
  • 注册一个外部的catalog，可以访问外部系统的数据或文件。
  • 执行SQL查询。
  • 注册一个用户自定义的function。
  • 将DataStream或DataSet转成Table。

一个查询中只能绑定一个指定的TableEnvironment，TableEnvironment可以通过来配置TableConfig来配置，通过TableConfig可以自定义查询优化以及translation的进程。

TableEnvironment执行过程如下：

  • TableEnvironment.sql()为调用入口；

  • Flink实现了FlinkPlannerImpl，执行parse(sql)，validate(sqlNode)，rel(sqlNode)操作；

  • 生成Table；

具体代码摘要如下

  1. package org.apache.Flink.table.api.internal;
  3. @Internal
  4. public class TableEnvironmentImpl implements TableEnvironment {
  5. 	private final CatalogManager catalogManager;
  6. 	private final ModuleManager moduleManager;
  7. 	private final OperationTreeBuilder operationTreeBuilder;
  8. 	private final List<ModifyOperation> bufferedModifyOperations = new ArrayList<>();
  10. 	protected final TableConfig tableConfig;
  11. 	protected final Executor execEnv;
  12. 	protected final FunctionCatalog functionCatalog;
  13. 	protected final Planner planner;
  14. 	protected final Parser parser;  
  15. }  
  17. // 在程序中打印类内容如下
  19. this = {StreamTableEnvironmentImpl@4701} 
  20.  functionCatalog = {FunctionCatalog@4702} 
  21.  scalaExecutionEnvironment = {StreamExecutionEnvironment@4703} 
  22.  planner = {StreamPlanner@4704} 
  23.   config = {TableConfig@4708} 
  24.   executor = {StreamExecutor@4709} 
  25.   PlannerBase.config = {TableConfig@4708} 
  26.   functionCatalog = {FunctionCatalog@4702} 
  27.   catalogManager = {CatalogManager@1250} 
  28.   isStreamingMode = true
  29.   plannerContext = {PlannerContext@4711} 
  30.   parser = {ParserImpl@4696} 
  31.  catalogManager = {CatalogManager@1250} 
  32.  moduleManager = {ModuleManager@4705} 
  33.  operationTreeBuilder = {OperationTreeBuilder@4706} 
  34.  bufferedModifyOperations = {ArrayList@4707}  size = 0
  35.  tableConfig = {TableConfig@4708} 
  36.  execEnv = {StreamExecutor@4709} 
  37.  TableEnvironmentImpl.functionCatalog = {FunctionCatalog@4702} 
  38.  TableEnvironmentImpl.planner = {StreamPlanner@4704} 
  39.  parser = {ParserImpl@4696} 
  40.  registration = {TableEnvironmentImpl$1@4710}

Catalog – 定义元数据和命名空间，包含 Schema（库），Table（表），RelDataType（类型信息）。

所有对数据库和表的元数据信息都存放在Flink CataLog内部目录结构中，其存放了Flink内部所有与Table相关的元数据信息，包括表结构信息/数据源信息等。

  1. // TableEnvironment里面包含一个CatalogManager
  2. public final class CatalogManager {
  3. 	// A map between names and catalogs.
  4. 	private Map<String, Catalog> catalogs;  
  5. } 
  7. // Catalog接口
  8. public interface Catalog {
  9.   ......
  10.   	default Optional<TableFactory> getTableFactory() {
  11. 		return Optional.empty();
  12. 	}
  13.   ......
  14. }   
  16. // 当数据来源是在程序里面自定义的时候，对应是GenericInMemoryCatalog
  17. public class GenericInMemoryCatalog extends AbstractCatalog {
  18. 	public static final String DEFAULT_DB = "default";
  19. 	private final Map<String, CatalogDatabase> databases;
  20. 	private final Map<ObjectPath, CatalogBaseTable> tables;
  21. 	private final Map<ObjectPath, CatalogFunction> functions;
  22. 	private final Map<ObjectPath, Map<CatalogPartitionSpec, CatalogPartition>> partitions;
  24. 	private final Map<ObjectPath, CatalogTableStatistics> tableStats;
  25. 	private final Map<ObjectPath, CatalogColumnStatistics> tableColumnStats;
  26. 	private final Map<ObjectPath, Map<CatalogPartitionSpec, CatalogTableStatistics>> partitionStats;
  27. 	private final Map<ObjectPath, Map<CatalogPartitionSpec, CatalogColumnStatistics>> partitionColumnStats;
  28. }  
  30. // 程序中调试的内容  
  32. catalogManager = {CatalogManager@4646} 
  33.  catalogs = {LinkedHashMap@4652}  size = 1
  34.   "default_catalog" -> {GenericInMemoryCatalog@4659} 
  35.    key = "default_catalog"
  36.     value = {char[15]@4668} 
  37.     hash = 552406043
  38.    value = {GenericInMemoryCatalog@4659} 
  39.     databases = {LinkedHashMap@4660}  size = 1
  40.     tables = {LinkedHashMap@4661}  size = 0
  41.     functions = {LinkedHashMap@4662}  size = 0
  42.     partitions = {LinkedHashMap@4663}  size = 0
  43.     tableStats = {LinkedHashMap@4664}  size = 0
  44.     tableColumnStats = {LinkedHashMap@4665}  size = 0
  45.     partitionStats = {LinkedHashMap@4666}  size = 0
  46.     partitionColumnStats = {LinkedHashMap@4667}  size = 0
  47.     catalogName = "default_catalog"
  48.     defaultDatabase = "default_database"
  49.  temporaryTables = {HashMap@4653}  size = 2
  50.  currentCatalogName = "default_catalog"
  51.  currentDatabaseName = "default_database"
  52.  builtInCatalogName = "default_catalog"

StreamPlanner是新的Blink Planner一种。

Flink Table 的新架构实现了查询处理器的插件化，社区完整保留原有 Flink Planner (Old Planner)，同时又引入了新的 Blink Planner，用户可以自行选择使用 Old Planner 还是 Blink Planner。

在模型上，Old Planner 没有考虑流计算作业和批处理作业的统一，针对流计算作业和批处理作业的实现不尽相同，在底层会分别翻译到 DataStream API 和 DataSet API 上。而 Blink Planner 将批数据集看作 bounded DataStream (有界流式数据) ，流计算作业和批处理作业最终都会翻译到 Transformation API 上。 在架构上，Blink Planner 针对批处理和流计算，分别实现了BatchPlanner 和 StreamPlanner ，两者共用了大部分代码，共享了很多优化逻辑。 Old Planner 针对批处理和流计算的代码实现的是完全独立的两套体系，基本没有实现代码和优化逻辑复用。

除了模型和架构上的优点外，Blink Planner 沉淀了许多实用功能，集中在三个方面：

  • Blink Planner 对代码生成机制做了改进、对部分算子进行了优化，提供了丰富实用的新功能，如维表 join、Top N、MiniBatch、流式去重、聚合场景的数据倾斜优化等新功能。
  • Blink Planner 的优化策略是基于公共子图的优化算法，包含了基于成本的优化（CBO）和基于规则的优化(CRO)两种策略，优化更为全面。同时，Blink Planner 支持从 catalog 中获取数据源的统计信息，这对CBO优化非常重要。
  • Blink Planner 提供了更多的内置函数，更标准的 SQL 支持，在 Flink 1.9 版本中已经完整支持 TPC-H ，对高阶的 TPC-DS 支持也计划在下一个版本实现。

具体对应代码来看，StreamPlanner体现在translateToPlan会调用到不同的 StreamOperator 生成系统上。

  1. class StreamPlanner(
  2.     executor: Executor,
  3.     config: TableConfig,
  4.     functionCatalog: FunctionCatalog,
  5.     catalogManager: CatalogManager)
  6.   extends PlannerBase(executor, config, functionCatalog, catalogManager, isStreamingMode = true) {
  8.   override protected def translateToPlan(
  9.       execNodes: util.List[ExecNode[_, _]]): util.List[Transformation[_]] = {
  10.     execNodes.map {
  11.       case node: StreamExecNode[_] => node.translateToPlan(this)
  12.       case _ =>
  13.         throw new TableException("Cannot generate DataStream due to an invalid logical plan. " +
  14.           "This is a bug and should not happen. Please file an issue.")
  15.     }
  16.   }
  17. }
  19. @Internal
  20. public final class StreamTableEnvironmentImpl extends TableEnvironmentImpl implements StreamTableEnvironment {
  22. 	private <T> DataStream<T> toDataStream(Table table, OutputConversionModifyOperation modifyOperation) {
  23.     
  24.     // 在转换回DataStream时候进行调用 planner 生成plan的操作。
  25.     
  26. 		List<Transformation<?>> transformations = planner.translate(Collections.singletonList(modifyOperation));
  28. 		Transformation<T> transformation = getTransformation(table, transformations);
  30. 		executionEnvironment.addOperator(transformation);
  31. 		return new DataStream<>(executionEnvironment, transformation);
  32. 	}
  33. }
  35. // 程序中调试打印的运行栈 
  37. translateToPlanInternal:85, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  38. translateToPlanInternal:39, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  39. translateToPlan:58, ExecNode$class (org.apache.Flink.table.planner.plan.nodes.exec)
  40. translateToPlan:39, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  41. translateToTransformation:184, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  42. translateToPlanInternal:153, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  43. translateToPlanInternal:48, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  44. translateToPlan:58, ExecNode$class (org.apache.Flink.table.planner.plan.nodes.exec)
  45. translateToPlan:48, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  46. apply:60, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.Flink.table.planner.delegation)
  47. apply:59, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.Flink.table.planner.delegation)
  48. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  49. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  50. foreach:891, Iterator$class (scala.collection)
  51. foreach:1334, AbstractIterator (scala.collection)
  52. foreach:72, IterableLike$class (scala.collection)
  53. foreach:54, AbstractIterable (scala.collection)
  54. map:234, TraversableLike$class (scala.collection)
  55. map:104, AbstractTraversable (scala.collection)
  56. translateToPlan:59, StreamPlanner (org.apache.Flink.table.planner.delegation)
  57. translate:153, PlannerBase (org.apache.Flink.table.planner.delegation)
  58. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  59. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  60. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  61. main:89, StreamSQLExample$ (spendreport)
  62. main:-1, StreamSQLExample (spendreport)

Flink实现了FlinkPlannerImpl，做为和Calcite 联系的桥梁，执行parse(sql)，validate(sqlNode)，rel(sqlNode)操作。

  1. class FlinkPlannerImpl(
  2.     config: FrameworkConfig,
  3.     catalogReaderSupplier: JFunction[JBoolean, CalciteCatalogReader],
  4.     typeFactory: FlinkTypeFactory,
  5.     cluster: RelOptCluster) {
  7.   val operatorTable: SqlOperatorTable = config.getOperatorTable
  8.   val parser: CalciteParser = new CalciteParser(config.getParserConfig)
  9.   val convertletTable: SqlRexConvertletTable = config.getConvertletTable
  10.   val sqlToRelConverterConfig: SqlToRelConverter.Config = config.getSqlToRelConverterConfig
  11. }
  13. // 这里会有使用 FlinkPlannerImpl
  14. public class ParserImpl implements Parser {
  15. 	private final CatalogManager catalogManager;
  16. 	private final Supplier<FlinkPlannerImpl> validatorSupplier;
  17. 	private final Supplier<CalciteParser> calciteParserSupplier;
  18.   
  19. 	@Override
  20. 	public List<Operation> parse(String statement) {
  21. 		CalciteParser parser = calciteParserSupplier.get();
  22.     // 这里会有使用 FlinkPlannerImpl
  23. 		FlinkPlannerImpl planner = validatorSupplier.get();
  24. 		// parse the sql query
  25. 		SqlNode parsed = parser.parse(statement);
  26. 		Operation operation = SqlToOperationConverter.convert(planner, catalogManager, parsed)
  27. 			.orElseThrow(() -> new TableException("Unsupported query: " + statement));
  28. 		return Collections.singletonList(operation);
  29. 	}  
  30. }
  32. // 程序中调试的内容  
  34. planner = {FlinkPlannerImpl@4659} 
  35.  config = {Frameworks$StdFrameworkConfig@4685} 
  36.  catalogReaderSupplier = {PlannerContext$lambda@4686} 
  37.  typeFactory = {FlinkTypeFactory@4687} 
  38.  cluster = {FlinkRelOptCluster@4688} 
  39.  operatorTable = {ChainedSqlOperatorTable@4689} 
  40.  parser = {CalciteParser@4690} 
  41.  convertletTable = {StandardConvertletTable@4691} 
  42.  sqlToRelConverterConfig = {SqlToRelConverter$ConfigImpl@4692} 
  43.  validator = null
  44.    
  45. // 程序调用栈之一   
  46.    
  47. validate:104, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  48. convert:127, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  49. parse:66, ParserImpl (org.apache.Flink.table.planner.delegation)
  50. sqlQuery:464, TableEnvironmentImpl (org.apache.Flink.table.api.internal)
  51. main:82, StreamSQLExample$ (spendreport)
  52. main:-1, StreamSQLExample (spendreport)
  53.   
  54. // 程序调用栈之二       
  55.   
  56. rel:135, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  57. toQueryOperation:522, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  58. convertSqlQuery:436, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  59. convert:154, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  60. parse:66, ParserImpl (org.apache.Flink.table.planner.delegation)
  61. sqlQuery:464, TableEnvironmentImpl (org.apache.Flink.table.api.internal)
  62. main:82, StreamSQLExample$ (spendreport)
  63. main:-1, StreamSQLExample (spendreport)

从代码中能看出，这就是个把各种相关操作和信息封装起来类而已，并不涉及太多实际逻辑。

  1. @Internal
  2. public class TableImpl implements Table {
  4. 	private static final AtomicInteger uniqueId = new AtomicInteger(0);
  6. 	private final TableEnvironment tableEnvironment;
  7. 	private final QueryOperation operationTree;
  8. 	private final OperationTreeBuilder operationTreeBuilder;
  9. 	private final LookupCallResolver lookupResolver;
  10.   
  11. 	private TableImpl joinInternal(
  12. 			Table right,
  13. 			Optional<Expression> joinPredicate,
  14. 			JoinType joinType) {
  15. 		verifyTableCompatible(right);
  17. 		return createTable(operationTreeBuilder.join(
  18. 			this.operationTree,
  19. 			right.getQueryOperation(),
  20. 			joinType,
  21. 			joinPredicate,
  22. 			false));
  23. 	}
  24. }
  26. // 程序中调试的内容 
  28. view = {TableImpl@4583} "UnnamedTable$0"
  29.  tableEnvironment = {StreamTableEnvironmentImpl@4580} 
  30.   functionCatalog = {FunctionCatalog@4646} 
  31.   scalaExecutionEnvironment = {StreamExecutionEnvironment@4579} 
  32.   planner = {StreamPlanner@4647} 
  33.   catalogManager = {CatalogManager@4644} 
  34.   moduleManager = {ModuleManager@4648} 
  35.   operationTreeBuilder = {OperationTreeBuilder@4649} 
  36.   bufferedModifyOperations = {ArrayList@4650}  size = 0
  37.   tableConfig = {TableConfig@4651} 
  38.   execEnv = {StreamExecutor@4652} 
  39.   TableEnvironmentImpl.functionCatalog = {FunctionCatalog@4646} 
  40.   TableEnvironmentImpl.planner = {StreamPlanner@4647} 
  41.   parser = {ParserImpl@4653} 
  42.   registration = {TableEnvironmentImpl$1@4654} 
  43.  operationTree = {ScalaDataStreamQueryOperation@4665} 
  44.   identifier = null
  45.   dataStream = {DataStreamSource@4676} 
  46.   fieldIndices = {int[2]@4677} 
  47.   tableSchema = {TableSchema@4678} "root\n |-- orderId: STRING\n |-- productName: STRING\n"
  48.  operationTreeBuilder = {OperationTreeBuilder@4649} 
  49.   config = {TableConfig@4651} 
  50.   functionCatalog = {FunctionCatalog@4646} 
  51.   tableReferenceLookup = {TableEnvironmentImpl$lambda@4668} 
  52.   lookupResolver = {LookupCallResolver@4669} 
  53.   projectionOperationFactory = {ProjectionOperationFactory@4670} 
  54.   sortOperationFactory = {SortOperationFactory@4671} 
  55.   calculatedTableFactory = {CalculatedTableFactory@4672} 
  56.   setOperationFactory = {SetOperationFactory@4673} 
  57.   aggregateOperationFactory = {AggregateOperationFactory@4674} 
  58.   joinOperationFactory = {JoinOperationFactory@4675} 
  59.  lookupResolver = {LookupCallResolver@4666} 
  60.   functionLookup = {FunctionCatalog@4646} 
  61.  tableName = "UnnamedTable$0"
  62.   value = {char[14]@4667} 
  63.   hash = 1355882650

这里对应前面脉络图，作用是生成了 SqlJoin 这样的 SqlNode

  1. // NOTE : 执行顺序是从上至下，" -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |  
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |

Calcite 使用 JavaCC 做 SQL 解析，JavaCC 根据 Calcite 中定义的 Parser.jj 文件，生成一系列的 java 代码，生成的 Java 代码会把 SQL 转换成 AST 的数据结构（这里是 SqlNode 类型）。

即：把 SQL 转换成为 AST （抽象语法树），在 Calcite 中用 SqlNode 来表示；

  1. package org.apache.Flink.table.planner.delegation;
  3. public class ParserImpl implements Parser {
  4. 	@Override
  5. 	public List<Operation> parse(String statement) {
  6. 		CalciteParser parser = calciteParserSupplier.get();
  7. 		FlinkPlannerImpl planner = validatorSupplier.get();
  8.     
  9. 		// parse the sql query
  10. 		SqlNode parsed = parser.parse(statement);
  12. 		Operation operation = SqlToOperationConverter.convert(planner, catalogManager, parsed)
  13. 			.orElseThrow(() -> new TableException("Unsupported query: " + statement));
  14. 		return Collections.singletonList(operation);
  15. 	}  
  16. }  
  18. // 打印出来解析之后 parsed 的内容，我们能看到 SqlNode 的基本格式。
  19.   
  20. parsed = {SqlBasicCall@4690} "SELECT *\nFROM `UnnamedTable$0`\nWHERE `amount` > 2\nUNION ALL\nSELECT *\nFROM `OrderB`\nWHERE `amount` < 2"
  21.  operator = {SqlSetOperator@4716} "UNION ALL"
  22.   all = true
  23.   name = "UNION ALL"
  24.   kind = {SqlKind@4742} "UNION"
  25.   leftPrec = 14
  26.   rightPrec = 15
  27.   returnTypeInference = {ReturnTypes$lambda@4743} 
  28.   operandTypeInference = null
  29.   operandTypeChecker = {SetopOperandTypeChecker@4744} 
  30.  operands = {SqlNode[2]@4717} 
  31.   0 = {SqlSelect@4746} "SELECT *\nFROM `UnnamedTable$0`\nWHERE `amount` > 2"
  32.   1 = {SqlSelect@4747} "SELECT *\nFROM `OrderB`\nWHERE `amount` < 2"
  33.  functionQuantifier = null
  34.  expanded = false
  35.  pos = {SqlParserPos@4719} "line 2, column 1"  
  36.     
  37. // 下面是调试相关Stack，可以帮助大家深入理解
  38.     
  39. SqlStmt:3208, FlinkSqlParserImpl (org.apache.Flink.sql.parser.impl)
  40. SqlStmtEof:3732, FlinkSqlParserImpl (org.apache.Flink.sql.parser.impl)
  41. parseSqlStmtEof:234, FlinkSqlParserImpl (org.apache.Flink.sql.parser.impl)
  42. parseQuery:160, SqlParser (org.apache.calcite.sql.parser)
  43. parseStmt:187, SqlParser (org.apache.calcite.sql.parser)
  44. parse:48, CalciteParser (org.apache.Flink.table.planner.calcite)
  45. parse:64, ParserImpl (org.apache.Flink.table.planner.delegation)
  46. sqlQuery:464, TableEnvironmentImpl (org.apache.Flink.table.api.internal)
  47. main:82, StreamSQLExample$ (spendreport)
  48. main:-1, StreamSQLExample (spendreport)
  49.   
  50. // 另一个参考 in FlinkSqlParserImpl.FromClause     
  51.     
  52. e = {SqlJoin@4709} "`Orders` AS `o`\nLEFT JOIN `Payment` AS `p` ON `o`.`orderId` = `p`.`orderId`"
  53.  left = {SqlBasicCall@4676} "`Orders` AS `o`"
  54.   operator = {SqlAsOperator@4752} "AS"
  55.   operands = {SqlNode[2]@4753} 
  56.   functionQuantifier = null
  57.   expanded = false
  58.   pos = {SqlParserPos@4755} "line 7, column 3"
  59.  natural = {SqlLiteral@4677} "FALSE"
  60.   typeName = {SqlTypeName@4775} "BOOLEAN"
  61.   value = {Boolean@4776} false
  62.   pos = {SqlParserPos@4777} "line 7, column 13"
  63.  joinType = {SqlLiteral@4678} "LEFT"
  64.   typeName = {SqlTypeName@4758} "SYMBOL"
  65.   value = {JoinType@4759} "LEFT"
  66.   pos = {SqlParserPos@4724} "line 7, column 26"
  67.  right = {SqlBasicCall@4679} "`Payment` AS `p`"
  68.   operator = {SqlAsOperator@4752} "AS"
  69.   operands = {SqlNode[2]@4763} 
  70.   functionQuantifier = null
  71.   expanded = false
  72.   pos = {SqlParserPos@4764} "line 7, column 31"
  73.  conditionType = {SqlLiteral@4680} "ON"
  74.   typeName = {SqlTypeName@4758} "SYMBOL"
  75.   value = {JoinConditionType@4771} "ON"
  76.   pos = {SqlParserPos@4772} "line 7, column 44"
  77.  condition = {SqlBasicCall@4681} "`o`.`orderId` = `p`.`orderId`"
  78.   operator = {SqlBinaryOperator@4766} "="
  79.   operands = {SqlNode[2]@4767} 
  80.   functionQuantifier = null
  81.   expanded = false
  82.   pos = {SqlParserPos@4768} "line 7, column 47"
  83.  pos = {SqlParserPos@4724} "line 7, column 26" 
  84.         
  85. // 下面是调试相关Stack，可以帮助大家深入理解
  86.     
  87. FromClause:10192, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  88. SqlSelect:5918, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  89. LeafQuery:630, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  90. LeafQueryOrExpr:15651, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  91. QueryOrExpr:15118, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  92. OrderedQueryOrExpr:504, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  93. SqlStmt:3693, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  94. SqlStmtEof:3732, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  95. parseSqlStmtEof:234, FlinkSqlParserImpl (org.apache.flink.sql.parser.impl)
  96. parseQuery:160, SqlParser (org.apache.calcite.sql.parser)
  97. parseStmt:187, SqlParser (org.apache.calcite.sql.parser)
  98. parse:48, CalciteParser (org.apache.flink.table.planner.calcite)
  99. parse:64, ParserImpl (org.apache.flink.table.planner.delegation)
  100. sqlQuery:464, TableEnvironmentImpl (org.apache.flink.table.api.internal)
  101. main:73, SimpleOuterJoin$ (spendreport)
  102. main:-1, SimpleOuterJoin (spendreport)

经过上面的第一步，会生成一个 SqlNode 对象，它是一个未经验证的抽象语法树，下面就进入了一个语法检查阶段，语法检查前需要知道元数据信息，这个检查会包括表名、字段名、函数名、数据类型的检查。

即：语法检查，根据元数据信息进行语法验证，验证之后还是用 SqlNode 表示 AST 语法树；

  1. package org.apache.Flink.table.planner.operations;
  3. public class SqlToOperationConverter {
  4. 	public static Optional<Operation> convert(
  5.     // 这里进行validate的调用
  6. 		final SqlNode validated = FlinkPlanner.validate(sqlNode);
  7. 		SqlToOperationConverter converter = new SqlToOperationConverter(FlinkPlanner, catalogManager);
  8. 	}    
  9. }
  10.     
  11. // 打印出来解析之后 validated 的内容。    
  12.     
  13. validated = {SqlBasicCall@4675} "SELECT `UnnamedTable$0`.`user`, `UnnamedTable$0`.`product`, `UnnamedTable$0`.`amount`\nFROM `default_catalog`.`default_database`.`UnnamedTable$0` AS `UnnamedTable$0`\nWHERE `UnnamedTable$0`.`amount` > 2\nUNION ALL\nSELECT `OrderB`.`user`, `OrderB`.`product`, `OrderB`.`amount`\nFROM `default_catalog`.`default_database`.`OrderB` AS `OrderB`\nWHERE `OrderB`.`amount` < 2"
  14.  operator = {SqlSetOperator@5000} "UNION ALL"
  15.   all = true
  16.   name = "UNION ALL"
  17.   kind = {SqlKind@5029} "UNION"
  18.   leftPrec = 14
  19.   rightPrec = 15
  20.   returnTypeInference = {ReturnTypes$lambda@5030} 
  21.   operandTypeInference = null
  22.   operandTypeChecker = {SetopOperandTypeChecker@5031} 
  23.  operands = {SqlNode[2]@5001} 
  24.   0 = {SqlSelect@4840} "SELECT `UnnamedTable$0`.`user`, `UnnamedTable$0`.`product`, `UnnamedTable$0`.`amount`\nFROM `default_catalog`.`default_database`.`UnnamedTable$0` AS `UnnamedTable$0`\nWHERE `UnnamedTable$0`.`amount` > 2"
  25.   1 = {SqlSelect@5026} "SELECT `OrderB`.`user`, `OrderB`.`product`, `OrderB`.`amount`\nFROM `default_catalog`.`default_database`.`OrderB` AS `OrderB`\nWHERE `OrderB`.`amount` < 2"
  26.  functionQuantifier = null
  27.  expanded = false
  28.  pos = {SqlParserPos@5003} "line 2, column 1"    
  29.     
  30. // 下面是调试相关Stack，可以帮助大家深入理解    
  31.     
  32. validate:81, AbstractNamespace (org.apache.calcite.sql.validate)
  33. validateNamespace:1008, SqlValidatorImpl (org.apache.calcite.sql.validate)
  34. validateQuery:968, SqlValidatorImpl (org.apache.calcite.sql.validate)
  35. validateCall:90, SqlSetOperator (org.apache.calcite.sql)
  36. validateCall:5304, SqlValidatorImpl (org.apache.calcite.sql.validate)
  37. validate:116, SqlCall (org.apache.calcite.sql)
  38. validateScopedExpression:943, SqlValidatorImpl (org.apache.calcite.sql.validate)
  39. validate:650, SqlValidatorImpl (org.apache.calcite.sql.validate)
  40. org$apache$Flink$table$planner$calcite$FlinkPlannerImpl$$validate:126, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  41. validate:105, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  42. convert:127, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  43. parse:66, ParserImpl (org.apache.Flink.table.planner.delegation)
  44. sqlQuery:464, TableEnvironmentImpl (org.apache.Flink.table.api.internal)
  45. main:82, StreamSQLExample$ (spendreport)
  46. main:-1, StreamSQLExample (spendreport)

脉络图中，这时候来到了

  1. // NOTE : 执行顺序是从上至下，" -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |      
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |     
  12. *     SqlToRelConverter.convertQuery // 语义分析，生成逻辑计划，作用是SqlNode–>RelNode
  13. *        |    
  14. *        |     
  15. *        +-----> LogicalProject (RelNode) // Abstract Syntax Tree，未优化的RelNode   
  16. *        |      
  17. *        |

经过第二步之后，这里的 SqlNode 就是经过语法校验的 SqlNode 树，接下来这一步就是将 SqlNode 转换成 RelNode/RexNode，也就是生成相应的逻辑计划（Logical Plan）

即：语义分析，根据 SqlNode及元信息构建 RelNode 树，也就是最初版本的逻辑计划（Logical Plan）；

根据这个已经生成的Flink的logical Plan，将它转换成calcite的logicalPlan，这样我们才能用到calcite强大的优化规则。

Flink由上往下依次调用各个节点的construct方法，将Flink节点转换成calcite的RelNode节点。真正的实现是在 convertQueryRecursive() 方法中完成的。

比如生成 LogicalProject 调用关系大概如下：

  1. createJoin:378, RelFactories$JoinFactoryImpl (org.apache.calcite.rel.core)
  2. createJoin:2520, SqlToRelConverter (org.apache.calcite.sql2rel)
  3. convertFrom:2111, SqlToRelConverter (org.apache.calcite.sql2rel)
  4. convertSelectImpl:646, SqlToRelConverter (org.apache.calcite.sql2rel)
  5. convertSelect:627, SqlToRelConverter (org.apache.calcite.sql2rel)
  6. convertQueryRecursive:3181, SqlToRelConverter (org.apache.calcite.sql2rel)
  7. convertQuery:563, SqlToRelConverter (org.apache.calcite.sql2rel)
  8. org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel:148, FlinkPlannerImpl (org.apache.flink.table.planner.calcite)
  9. rel:135, FlinkPlannerImpl (org.apache.flink.table.planner.calcite)
  10. toQueryOperation:522, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  11. convertSqlQuery:436, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  12. convert:154, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  13. parse:66, ParserImpl (org.apache.flink.table.planner.delegation)
  14. sqlQuery:464, TableEnvironmentImpl (org.apache.flink.table.api.internal)
  15. main:73, SimpleOuterJoin$ (spendreport)
  16. main:-1, SimpleOuterJoin (spendreport)

具体详细源码如下：

  1. SqlToRelConverter 中的 convertQuery() 将 SqlNode 转换为 RelRoot
  2.   
  3. public class SqlToRelConverter {  
  4.     public RelRoot convertQuery(SqlNode query, boolean needsValidation, boolean top) {
  5.         if (needsValidation) {
  6.             query = this.validator.validate(query);
  7.         }
  8.               	             RelMetadataQuery.THREAD_PROVIDERS.set(JaninoRelMetadataProvider.of(this.cluster.getMetadataProvider()));
  9.         RelNode result = this.convertQueryRecursive(query, top, (RelDataType)null).rel;
  10.         if (top && isStream(query)) {
  11.             result = new LogicalDelta(this.cluster, ((RelNode)result).getTraitSet(), (RelNode)result);
  12.         }
  14.         RelCollation collation = RelCollations.EMPTY;
  15.         if (!query.isA(SqlKind.DML) && isOrdered(query)) {
  16.             collation = this.requiredCollation((RelNode)result);
  17.         }
  19.         this.checkConvertedType(query, (RelNode)result);
  21.         RelDataType validatedRowType = this.validator.getValidatedNodeType(query);
  22.       
  23.         // 这里设定了Root
  24.         return RelRoot.of((RelNode)result, validatedRowType, query.getKind()).withCollation(collation);
  25.     }
  26. }
  28. // 在这里打印
  29. toQueryOperation:523, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  31. // 得到如下内容，可以看到一个RelRoot的真实结构
  32.   
  33. relational = {RelRoot@5248} "Root {kind: UNION, rel: LogicalUnion#6, rowType: RecordType(BIGINT user, VARCHAR(2147483647) product, INTEGER amount), fields: [<0, user>, <1, product>, <2, amount>], collation: []}"
  34.  rel = {LogicalUnion@5227} "LogicalUnion#6"
  35.   inputs = {RegularImmutableList@5272}  size = 2
  36.   kind = {SqlKind@5029} "UNION"
  37.   all = true
  38.   desc = "LogicalUnion#6"
  39.   rowType = {RelRecordType@5238} "RecordType(BIGINT user, VARCHAR(2147483647) product, INTEGER amount)"
  40.   digest = "LogicalUnion#6"
  41.   cluster = {FlinkRelOptCluster@4800} 
  42.   id = 6
  43.   traitSet = {RelTraitSet@5273}  size = 5
  44.  validatedRowType = {RelRecordType@5238} "RecordType(BIGINT user, VARCHAR(2147483647) product, INTEGER amount)"
  45.   kind = {StructKind@5268} "FULLY_QUALIFIED"
  46.   nullable = false
  47.   fieldList = {RegularImmutableList@5269}  size = 3
  48.   digest = "RecordType(BIGINT user, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" product, INTEGER amount) NOT NULL"
  49.  kind = {SqlKind@5029} "UNION"
  50.   lowerName = "union"
  51.   sql = "UNION"
  52.   name = "UNION"
  53.   ordinal = 18
  54.  fields = {RegularImmutableList@5254}  size = 3
  55.   {Integer@5261} 0 -> "user"
  56.   {Integer@5263} 1 -> "product"
  57.   {Integer@5265} 2 -> "amount"
  58.  collation = {RelCollationImpl@5237} "[]"
  59.   fieldCollations = {RegularImmutableList@5256}  size = 0
  61. // 调用栈内容
  62.     
  63. convertQuery:561, SqlToRelConverter (org.apache.calcite.sql2rel)
  64. org$apache$Flink$table$planner$calcite$FlinkPlannerImpl$$rel:148, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  65. rel:135, FlinkPlannerImpl (org.apache.Flink.table.planner.calcite)
  66. toQueryOperation:522, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  67. convertSqlQuery:436, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  68. convert:154, SqlToOperationConverter (org.apache.Flink.table.planner.operations)
  69. parse:66, ParserImpl (org.apache.Flink.table.planner.delegation)
  70. sqlQuery:464, TableEnvironmentImpl (org.apache.Flink.table.api.internal)
  71. main:82, StreamSQLExample$ (spendreport)
  72. main:-1, StreamSQLExample (spendreport)
  73.   
  74. // 再次举例，生成了LogicalProject 
  75.   
  76. bb = {SqlToRelConverter$Blackboard@4978} 
  77.  scope = {SelectScope@4977} 
  78.  nameToNodeMap = null
  79.  root = {LogicalProject@5100} "LogicalProject#4"
  80.   exps = {RegularImmutableList@5105}  size = 3
  81.   input = {LogicalJoin@5106} "LogicalJoin#3"
  82.   desc = "LogicalProject#4"
  83.   rowType = {RelRecordType@5107} "RecordType(VARCHAR(2147483647) orderId, VARCHAR(2147483647) productName, VARCHAR(2147483647) payType)"
  84.   digest = "LogicalProject#4"
  85.   cluster = {FlinkRelOptCluster@4949} 
  86.   id = 4
  87.   traitSet = {RelTraitSet@5108}  size = 5
  88.  inputs = {Collections$SingletonList@5111}  size = 1
  89.  mapCorrelateToRex = {HashMap@5112}  size = 0
  90.  isPatternVarRef = false
  91.  cursors = {ArrayList@5113}  size = 0
  92.  subQueryList = {LinkedHashSet@5114}  size = 0
  93.  agg = null
  94.  window = null
  95.  mapRootRelToFieldProjection = {HashMap@5115}  size = 0
  96.  columnMonotonicities = {ArrayList@5116}  size = 3
  97.  systemFieldList = {ArrayList@5117}  size = 0
  98.  top = true
  99.  initializerExpressionFactory = {NullInitializerExpressionFactory@5118} 
  100.  this$0 = {SqlToRelConverter@4926} 
  102. // 举例，LogicalProject是在这里生成的。
  104.    protected void convertFrom(SqlToRelConverter.Blackboard bb, SqlNode from) {
  105.             case JOIN:
  106.                 RelNode joinRel = this.createJoin(fromBlackboard, leftRel, rightRel, conditionExp, convertedJoinType);
  107.                 bb.setRoot(joinRel, false);  
  108.    }
  109.   
  110. // 相关调用栈
  112. createJoin:378, RelFactories$JoinFactoryImpl (org.apache.calcite.rel.core)
  113. createJoin:2520, SqlToRelConverter (org.apache.calcite.sql2rel)
  114. convertFrom:2111, SqlToRelConverter (org.apache.calcite.sql2rel)
  115. convertSelectImpl:646, SqlToRelConverter (org.apache.calcite.sql2rel)
  116. convertSelect:627, SqlToRelConverter (org.apache.calcite.sql2rel)
  117. convertQueryRecursive:3181, SqlToRelConverter (org.apache.calcite.sql2rel)
  118. convertQuery:563, SqlToRelConverter (org.apache.calcite.sql2rel)
  119. org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel:148, FlinkPlannerImpl (org.apache.flink.table.planner.calcite)
  120. rel:135, FlinkPlannerImpl (org.apache.flink.table.planner.calcite)
  121. toQueryOperation:522, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  122. convertSqlQuery:436, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  123. convert:154, SqlToOperationConverter (org.apache.flink.table.planner.operations)
  124. parse:66, ParserImpl (org.apache.flink.table.planner.delegation)
  125. sqlQuery:464, TableEnvironmentImpl (org.apache.flink.table.api.internal)
  126. main:73, SimpleOuterJoin$ (spendreport)
  127. main:-1, SimpleOuterJoin (spendreport)

这时候，脉络图到了这里

  1. // NOTE : 执行顺序是从上至下，" -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |      
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |     
  12. *     SqlToRelConverter.convertQuery // 语义分析，生成逻辑计划，作用是SqlNode–>RelNode
  13. *        |    
  14. *        |     
  15. *        +-----> LogicalProject (RelNode) // Abstract Syntax Tree，未优化的RelNode   
  16. *        |      
  17. *        |     
  18. *    FlinkLogicalJoinConverter (RelOptRule) // Flink定制的优化rules      
  19. *    VolcanoRuleCall.onMatch // 基于Flink定制的一些优化rules去优化 Logical Plan 
  20. *        | 
  21. *        |   
  22. *        +-----> FlinkLogicalJoin (RelNode)  // Optimized Logical Plan，逻辑执行计划
  23. *        |  
  24. *        |    
  25. *    StreamExecJoinRule (RelOptRule) // Rule that converts FlinkLogicalJoin without window bounds in join condition to StreamExecJoin     
  26. *    VolcanoRuleCall.onMatch // 基于Flink rules将optimized LogicalPlan转成Flink物理执行计划
  27. *        |       
  28. *        |   
  29. *        +-----> StreamExecJoin (FlinkRelNode) // Stream physical RelNode，物理执行计划
  30. *        |      
  31. *        |

第四阶段，也就是 Calcite 的核心所在。

即：逻辑计划优化，优化器的核心，根据前面生成的逻辑计划按照相应的规则（Rule）进行优化；

Flink的这部分实现统一封装在optimize方法里头。这部分涉及到多个阶段，每个阶段都是用Rule对逻辑计划进行优化和改进。

在 Calcite 架构中，最核心地方就是 Optimizer，也就是优化器，一个 Optimization Engine 包含三个组成部分：

  • rules：也就是匹配规则，Calcite 内置上百种 Rules 来优化 relational expression，当然也支持自定义 rules；
  • metadata providers：主要是向优化器提供信息，这些信息会有助于指导优化器向着目标（减少整体 cost）进行优化，信息可以包括行数、table 哪一列是唯一列等，也包括计算 RelNode 树中执行 subexpression cost 的函数；
  • planner engines：它的主要目标是进行触发 rules 来达到指定目标，比如像 cost-based optimizer（CBO）的目标是减少cost（Cost 包括处理的数据行数、CPU cost、IO cost 等）。

优化器的作用是将解析器生成的关系代数表达式转换成执行计划，供执行引擎执行，在这个过程中，会应用一些规则优化，以帮助生成更高效的执行计划。优化器进行优化的地方如过滤条件的下压（push down），在进行 join 操作前，先进行 filter 操作，这样的话就不需要在 join 时进行全量 join，减少参与 join 的数据量等。

Calcite 中 RelOptPlanner 是 Calcite 中优化器的基类。Calcite 中关于优化器提供了两种实现：

  • HepPlanner：就是基于规则优化RBO 的实现，它是一个启发式的优化器，按照规则进行匹配，直到达到次数限制（match 次数限制）或者遍历一遍后不再出现 rule match 的情况才算完成；
  • VolcanoPlanner：就是基于成本优化CBO 的实现，它会一直迭代 rules，直到找到 cost 最小的 paln。

基于代价的优化器(Cost-Based Optimizer，CBO) 是根据优化规则对关系表达式进行转换。这里的转换是说一个关系表达式经过优化规则后会生成另外一个关系表达式，同时原有表达式也会保留，经过一系列转换后会生成多个执行计划，然后 CBO 会根据统计信息和代价模型 (Cost Model) 计算每个执行计划的 Cost，从中挑选 Cost 最小的执行计划。

由上可知，CBO 中有两个依赖：统计信息和代价模型。统计信息的准确与否、代价模型的合理与否都会影响 CBO 选择最优计划。 从上述描述可知，CBO 是优于 RBO 的，原因是 RBO 是一种只认规则，对数据不敏感的呆板的优化器，而在实际过程中，数据往往是有变化的，通过 RBO 生成的执行计划很有可能不是最优的。事实上目前各大数据库和大数据计算引擎都倾向于使用 CBO，但是对于流式计算引擎来说，使用 CBO 还是有很大难度的，因为并不能提前预知数据量等信息，这会极大地影响优化效果，CBO 主要还是应用在离线的场景。

VolcanoPlanner就是 CBO 的实现，它会一直迭代 rules，直到找到 cost 最小的 paln。其部分相关概念如下：

  • RelSet 描述一组等价 Relation Expression，所有的 RelNode 会记录在 rels 中；
  • RelSubset 描述一组物理属性相同的等价 Relation Expression，即它们具有相同的 Physical Properties；每个 RelSubset 都会记录其所属的 RelSet；RelSubset 继承自 AbstractRelNode，它也是一种 RelNode，物理属性记录在其成员变量 traitSet 中。每个 RelSubset 都将会记录其最佳 plan（best）和最佳 plan 的 cost（bestCost）信息。
  • RuleMatch 是对 Rule 和 RelSubset 关系的一个抽象，它会记录这两者的信息。
  • importance 决定了在进行 Rule 优化时 Rule 应用的顺序，它是一个相对概念，在 VolcanoPlanner 中有两个 importance，分别是 RelSubset 和 RuleMatch 的 importance

在应用 VolcanoPlanner 时，整体分为以下四步：

  1. 初始化 VolcanoPlanner，并向 Rule Match Queue 中添加相应的 Rule Match（包括 ConverterRule）；
  2. 对 RelNode 做等价转换：应用 Rule Match 对 plan graph 做 transformation 优化（Rule specifies an Operator sub-graph to match and logic to generate equivalent better sub-graph）；这里只是改变其物理属性（Convention）；
  3. 通过 VolcanoPlanner 的 setRoot() 方法注册相应的 RelNode，并进行相应的初始化操作；
  4. 通过动态规划算法进行相应的迭代，直到 cost 不再变化或者 Rule Match Queue 中 rule match 已经全部应用完成；这样找到 cost 最小的 plan；Rule Match 的 importance 依赖于 RelNode 的 cost 和深度。

下面通过这个 示例 来详细看下 VolcanoPlanner 内部的实现逻辑。

  1. //1. 初始化 VolcanoPlanner 对象，并添加相应的 Rule
  2. VolcanoPlanner planner = new VolcanoPlanner();
  3. planner.addRelTraitDef(ConventionTraitDef.INSTANCE);
  4. planner.addRelTraitDef(RelDistributionTraitDef.INSTANCE);
  5. // 添加相应的 rule
  6.   planner.addRule(FilterJoinRule.FilterIntoJoinRule.FILTER_ON_JOIN);
  7. planner.addRule(ReduceExpressionsRule.PROJECT_INSTANCE);
  8. planner.addRule(PruneEmptyRules.PROJECT_INSTANCE);
  9. // 添加相应的 ConverterRule
  10. planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
  11. planner.addRule(EnumerableRules.ENUMERABLE_SORT_RULE);
  12. planner.addRule(EnumerableRules.ENUMERABLE_VALUES_RULE);
  13. planner.addRule(EnumerableRules.ENUMERABLE_PROJECT_RULE);
  14. planner.addRule(EnumerableRules.ENUMERABLE_FILTER_RULE);
  15. //2. Changes a relational expression to an equivalent one with a different set of traits.
  16. RelTraitSet desiredTraits =
  17.     relNode.getCluster().traitSet().replace(EnumerableConvention.INSTANCE);
  18. relNode = planner.changeTraits(relNode, desiredTraits);
  19. //3. 通过 VolcanoPlanner 的 setRoot 方法注册相应的 RelNode，并进行相应的初始化操作
  20. planner.setRoot(relNode);
  21. //4. 通过动态规划算法找到 cost 最小的 plan
  22. relNode = planner.findBestExp();

Flink 中相关代码如下：

  1. public PlannerContext(
  2. 			TableConfig tableConfig,
  3. 			FunctionCatalog functionCatalog,
  4. 			CatalogManager catalogManager,
  5. 			CalciteSchema rootSchema,
  6. 			List<RelTraitDef> traitDefs) {
  7. 		this.tableConfig = tableConfig;
  9. 		this.context = new FlinkContextImpl(
  10. 				tableConfig,
  11. 				functionCatalog,
  12. 				catalogManager,
  13. 				this::createSqlExprToRexConverter);
  15. 		this.rootSchema = rootSchema;
  16. 		this.traitDefs = traitDefs;
  17. 		// Make a framework config to initialize the RelOptCluster instance,
  18. 		// caution that we can only use the attributes that can not be overwrite/configured
  19. 		// by user.
  20. 		this.frameworkConfig = createFrameworkConfig();
  22.     // 这里使用了VolcanoPlanner
  23. 		RelOptPlanner planner = new VolcanoPlanner(frameworkConfig.getCostFactory(), frameworkConfig.getContext());
  24. 		planner.setExecutor(frameworkConfig.getExecutor());
  25. 		for (RelTraitDef traitDef : frameworkConfig.getTraitDefs()) {
  26. 			planner.addRelTraitDef(traitDef);
  27. 		}
  28. 		this.cluster = FlinkRelOptClusterFactory.create(planner, new RexBuilder(typeFactory));
  29. 	}
  32. //初始化
  33. <init>:119, PlannerContext (org.apache.Flink.table.planner.delegation)
  34. <init>:86, PlannerBase (org.apache.Flink.table.planner.delegation)
  35. <init>:44, StreamPlanner (org.apache.Flink.table.planner.delegation)
  36. create:50, BlinkPlannerFactory (org.apache.Flink.table.planner.delegation)
  37. create:325, StreamTableEnvironmentImpl$ (org.apache.Flink.table.api.scala.internal)
  38. create:425, StreamTableEnvironment$ (org.apache.Flink.table.api.scala)
  39. main:56, StreamSQLExample$ (spendreport)
  40. main:-1, StreamSQLExample (spendreport)
  43. class FlinkVolcanoProgram[OC <: FlinkOptimizeContext] extends FlinkRuleSetProgram[OC] {
  45.  override def optimize(root: RelNode, context: OC): RelNode = {
  46.     val targetTraits = root.getTraitSet.plusAll(requiredOutputTraits.get).simplify()
  47.     // VolcanoPlanner limits that the planer a RelNode tree belongs to and
  48.     // the VolcanoPlanner used to optimize the RelNode tree should be same instance.
  49.     // see: VolcanoPlanner#registerImpl
  50.     // here, use the planner in cluster directly
  51.       
  52.     // 这里也使用了VolcanoPlanner 
  53.     val planner = root.getCluster.getPlanner.asInstanceOf[VolcanoPlanner]
  54.     val optProgram = Programs.ofRules(rules)
  55.   }
  56. }      
  57.   
  58. // 其调用栈
  60. optimize:60, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  61. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  62. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  63. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  64. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  65. foreach:891, Iterator$class (scala.collection)
  66. foreach:1334, AbstractIterator (scala.collection)
  67. foreach:72, IterableLike$class (scala.collection)
  68. foreach:54, AbstractIterable (scala.collection)
  69. foldLeft:157, TraversableOnce$class (scala.collection)
  70. foldLeft:104, AbstractTraversable (scala.collection)
  71. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  72. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  73. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  74. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  75. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  76. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  77. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  78. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  79. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  80. main:89, StreamSQLExample$ (spendreport)
  81. main:-1, StreamSQLExample (spendreport)
  83. // 下面全部是 VolcanoPlanner 相关代码和调用栈 
  84.   
  85. // VolcanoPlanner添加Rule，筛选出来的优化规则会封装成VolcanoRuleMatch，然后扔到RuleQueue里，而这个RuleQueue正是接下来执行动态规划算法要用到的核心类。   
  86. public class VolcanoPlanner extends AbstractRelOptPlanner {
  87.       public boolean addRule(RelOptRule rule) {
  88.         ......
  89.       }
  90. }    
  91.         
  92. addRule:438, VolcanoPlanner (org.apache.calcite.plan.volcano)
  93. run:315, Programs$RuleSetProgram (org.apache.calcite.tools)
  94. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  95. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  96. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  97. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  98. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  99. foreach:891, Iterator$class (scala.collection)
  100. foreach:1334, AbstractIterator (scala.collection)
  101. foreach:72, IterableLike$class (scala.collection)
  102. foreach:54, AbstractIterable (scala.collection)
  103. foldLeft:157, TraversableOnce$class (scala.collection)
  104. foldLeft:104, AbstractTraversable (scala.collection)
  105. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  106. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  107. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  108. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  109. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  110. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  111. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  112. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  113. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  114. main:89, StreamSQLExample$ (spendreport)
  115. main:-1, StreamSQLExample (spendreport)        
  116.  
  117. // VolcanoPlanner修改Traits   
  118. public class VolcanoPlanner extends AbstractRelOptPlanner {
  119.     public RelNode changeTraits(RelNode rel, RelTraitSet toTraits) {
  120.         assert !rel.getTraitSet().equals(toTraits);
  122.         assert toTraits.allSimple();
  124.         RelSubset rel2 = this.ensureRegistered(rel, (RelNode)null);
  125.         return rel2.getTraitSet().equals(toTraits) ? rel2 : rel2.set.getOrCreateSubset(rel.getCluster(), toTraits.simplify());
  126.     }
  127. }
  129. changeTraits:529, VolcanoPlanner (org.apache.calcite.plan.volcano)
  130. run:324, Programs$RuleSetProgram (org.apache.calcite.tools)
  131. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  132. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  133. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  134. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  135. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  136. foreach:891, Iterator$class (scala.collection)
  137. foreach:1334, AbstractIterator (scala.collection)
  138. foreach:72, IterableLike$class (scala.collection)
  139. foreach:54, AbstractIterable (scala.collection)
  140. foldLeft:157, TraversableOnce$class (scala.collection)
  141. foldLeft:104, AbstractTraversable (scala.collection)
  142. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  143. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  144. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  145. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  146. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  147. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  148. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  149. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  150. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  151. main:89, StreamSQLExample$ (spendreport)
  152. main:-1, StreamSQLExample (spendreport)
  153.   
  154. // VolcanoPlanner设定Root 
  155. public class VolcanoPlanner extends AbstractRelOptPlanner {  
  156.     public void setRoot(RelNode rel) {
  157.         this.registerMetadataRels();
  158.         this.root = this.registerImpl(rel, (RelSet)null);
  159.         if (this.originalRoot == null) {
  160.             this.originalRoot = rel;
  161.         }
  163.         this.ruleQueue.recompute(this.root);
  164.         this.ensureRootConverters();
  165.     }
  166. }
  168. setRoot:294, VolcanoPlanner (org.apache.calcite.plan.volcano)
  169. run:326, Programs$RuleSetProgram (org.apache.calcite.tools)
  170. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  171. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  172. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  173. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  174. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  175. foreach:891, Iterator$class (scala.collection)
  176. foreach:1334, AbstractIterator (scala.collection)
  177. foreach:72, IterableLike$class (scala.collection)
  178. foreach:54, AbstractIterable (scala.collection)
  179. foldLeft:157, TraversableOnce$class (scala.collection)
  180. foldLeft:104, AbstractTraversable (scala.collection)
  181. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  182. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  183. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  184. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  185. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  186. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  187. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  188. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  189. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  190. main:89, StreamSQLExample$ (spendreport)
  191. main:-1, StreamSQLExample (spendreport)
  192.   
  193. // VolcanoPlanner找到最小cost，本质上就是一个动态规划算法的实现。
  194.   
  195. public class VolcanoPlanner extends AbstractRelOptPlanner {
  196.     public RelNode findBestExp() {
  197.         this.ensureRootConverters();
  198.         this.registerMaterializations();
  199.         int cumulativeTicks = 0;
  200.         VolcanoPlannerPhase[] var2 = VolcanoPlannerPhase.values();
  201.         int var3 = var2.length;
  203.         for(int var4 = 0; var4 < var3; ++var4) {
  204.             VolcanoPlannerPhase phase = var2[var4];
  205.             this.setInitialImportance();
  206.             RelOptCost targetCost = this.costFactory.makeHugeCost();
  207.             int tick = 0;
  208.             int firstFiniteTick = -1;
  209.             int splitCount = 0;
  210.             int giveUpTick = 2147483647;
  212.             while(true) {
  213.                 ++tick;
  214.                 ++cumulativeTicks;
  215.                 if (this.root.bestCost.isLe(targetCost)) {
  216.                     if (firstFiniteTick < 0) {
  217.                         firstFiniteTick = cumulativeTicks;
  218.                         this.clearImportanceBoost();
  219.                     }
  221.                     if (!this.ambitious) {
  222.                         break;
  223.                     }
  225.                     targetCost = this.root.bestCost.multiplyBy(0.9D);
  226.                     ++splitCount;
  227.                     if (this.impatient) {
  228.                         if (firstFiniteTick < 10) {
  229.                             giveUpTick = cumulativeTicks + 25;
  230.                         } else {
  231.                             giveUpTick = cumulativeTicks + Math.max(firstFiniteTick / 10, 25);
  232.                         }
  233.                     }
  234.                 } else {
  235.                     if (cumulativeTicks > giveUpTick) {
  236.                         break;
  237.                     }
  239.                     if (this.root.bestCost.isInfinite() && tick % 10 == 0) {
  240.                         this.injectImportanceBoost();
  241.                     }
  242.                 }
  244.                 VolcanoRuleMatch match = this.ruleQueue.popMatch(phase);
  245.                 if (match == null) {
  246.                     break;
  247.                 }
  249.                 assert match.getRule().matches(match);
  251.                 match.onMatch();
  252.                 this.root = this.canonize(this.root);
  253.             }
  255.             this.ruleQueue.phaseCompleted(phase);
  256.         }
  258.         RelNode cheapest = this.root.buildCheapestPlan(this);
  260.         return cheapest;
  261.     }
  262. }
  264. // VolcanoPlanner得到的Flink逻辑节点 cheapest，就是最终选择的结点
  266. cheapest = {FlinkLogicalUnion@6487} "FlinkLogicalUnion#443"
  267.  cluster = {FlinkRelOptCluster@6224} 
  268.  inputs = {RegularImmutableList@6493}  size = 2
  269.   0 = {FlinkLogicalCalc@6498} "FlinkLogicalCalc#441"
  270.    cluster = {FlinkRelOptCluster@6224} 
  271.    calcProgram = {RexProgram@6509} "(expr#0..2=[{inputs}], expr#3=[2], expr#4=[>($t2, $t3)], proj#0..2=[{exprs}], $condition=[$t4])"
  272.    program = {RexProgram@6509} "(expr#0..2=[{inputs}], expr#3=[2], expr#4=[>($t2, $t3)], proj#0..2=[{exprs}], $condition=[$t4])"
  273.    input = {FlinkLogicalDataStreamTableScan@6510} "rel#437:FlinkLogicalDataStreamTableScan.LOGICAL.any.None: 0.false.UNKNOWN(table=[default_catalog, default_database, UnnamedTable$0])"
  274.    desc = "FlinkLogicalCalc#441"
  275.    rowType = {RelRecordType@6504} "RecordType(BIGINT user, VARCHAR(2147483647) product, INTEGER amount)"
  276.    digest = "FlinkLogicalCalc#441"
  277.    AbstractRelNode.cluster = {FlinkRelOptCluster@6224} 
  278.    id = 441
  279.    traitSet = {RelTraitSet@5942}  size = 5
  280.   1 = {FlinkLogicalCalc@6499} "FlinkLogicalCalc#442"
  281.    cluster = {FlinkRelOptCluster@6224} 
  282.    calcProgram = {RexProgram@6502} "(expr#0..2=[{inputs}], expr#3=[2], expr#4=[<($t2, $t3)], proj#0..2=[{exprs}], $condition=[$t4])"
  283.    program = {RexProgram@6502} "(expr#0..2=[{inputs}], expr#3=[2], expr#4=[<($t2, $t3)], proj#0..2=[{exprs}], $condition=[$t4])"
  284.    input = {FlinkLogicalDataStreamTableScan@6503} "rel#435:FlinkLogicalDataStreamTableScan.LOGICAL.any.None: 0.false.UNKNOWN(table=[default_catalog, default_database, OrderB])"
  285.    desc = "FlinkLogicalCalc#442"
  286.    rowType = {RelRecordType@6504} "RecordType(BIGINT user, VARCHAR(2147483647) product, INTEGER amount)"
  287.    digest = "FlinkLogicalCalc#442"
  288.    AbstractRelNode.cluster = {FlinkRelOptCluster@6224} 
  289.    id = 442
  290.    traitSet = {RelTraitSet@5942}  size = 5
  291.  kind = {SqlKind@6494} "UNION"
  292.   lowerName = "union"
  293.   sql = "UNION"
  294.   name = "UNION"
  295.   ordinal = 18
  296.  all = true
  297.  desc = "FlinkLogicalUnion#443"
  298.  rowType = null
  299.  digest = "FlinkLogicalUnion#443"
  300.  AbstractRelNode.cluster = {FlinkRelOptCluster@6224} 
  301.  id = 443
  302.  traitSet = {RelTraitSet@5942}  size = 5
  305. findBestExp:572, VolcanoPlanner (org.apache.calcite.plan.volcano)
  306. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  307. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  308. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  309. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  310. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  311. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  312. foreach:891, Iterator$class (scala.collection)
  313. foreach:1334, AbstractIterator (scala.collection)
  314. foreach:72, IterableLike$class (scala.collection)
  315. foreach:54, AbstractIterable (scala.collection)
  316. foldLeft:157, TraversableOnce$class (scala.collection)
  317. foldLeft:104, AbstractTraversable (scala.collection)
  318. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  319. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  320. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  321. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  322. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  323. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  324. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  325. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  326. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  327. main:89, StreamSQLExample$ (spendreport)
  328. main:-1, StreamSQLExample (spendreport)

以下是Join的优化 。

  1. class FlinkLogicalJoin(
  2.     cluster: RelOptCluster,
  3.     traitSet: RelTraitSet,
  4.     left: RelNode,
  5.     right: RelNode,
  6.     condition: RexNode,
  7.     joinType: JoinRelType)
  8.   extends FlinkLogicalJoinBase(
  10.   override def convert(rel: RelNode): RelNode = {
  11.     val join = rel.asInstanceOf[LogicalJoin]
  12.     val traitSet = rel.getTraitSet.replace(FlinkConventions.LOGICAL)
  13.     val newLeft = RelOptRule.convert(join.getLeft, FlinkConventions.LOGICAL)
  14.     val newRight = RelOptRule.convert(join.getRight, FlinkConventions.LOGICAL)
  16.     new FlinkLogicalJoin(
  17.       rel.getCluster,
  18.       traitSet,
  19.       newLeft,
  20.       newRight,
  21.       join.getCondition,
  22.       join.getJoinType)
  23.   }
  24. }  
  26. call = {VolcanoRuleMatch@6191} "rule [FlinkLogicalJoinConverter(in:NONE,out:LOGICAL)] rels [rel#100:LogicalJoin.NONE.any.None: 0.false.UNKNOWN(left=RelSubset#98,right=RelSubset#99,condition==($0, $2),joinType=left)]"
  27.  targetSet = {RelSet@6193} 
  28.  targetSubset = null
  29.  digest = "rule [FlinkLogicalJoinConverter(in:NONE,out:LOGICAL)] rels [rel#100:LogicalJoin.NONE.any.None: 0.false.UNKNOWN(left=RelSubset#98,right=RelSubset#99,condition==($0, $2),joinType=left)]"
  30.  cachedImportance = 0.8019000000000001
  31.  volcanoPlanner = {VolcanoPlanner@6194} 
  32.  generatedRelList = null
  33.  id = 71
  34.  operand0 = {RelOptRule$ConverterRelOptRuleOperand@6186} 
  35.   parent = null
  36.   rule = {FlinkLogicalJoinConverter@6179} "FlinkLogicalJoinConverter(in:NONE,out:LOGICAL)"
  37.   predicate = {ConverterRule$lambda@6246} 
  38.   solveOrder = {int[1]@6247} 
  39.   ordinalInParent = 0
  40.   ordinalInRule = 0
  41.   trait = {Convention$Impl@6184} "NONE"
  42.   clazz = {Class@5010} "class org.apache.calcite.rel.logical.LogicalJoin"
  43.   children = {RegularImmutableList@6230}  size = 0
  44.   childPolicy = {RelOptRuleOperandChildPolicy@6248} "ANY"
  45.  nodeInputs = {RegularImmutableBiMap@6195}  size = 0
  46.  rule = {FlinkLogicalJoinConverter@6179} "FlinkLogicalJoinConverter(in:NONE,out:LOGICAL)"
  47.  rels = {RelNode[1]@6196} 
  48.   0 = {LogicalJoin@6181} "rel#100:LogicalJoin.NONE.any.None: 0.false.UNKNOWN(left=RelSubset#98,right=RelSubset#99,condition==($0, $2),joinType=left)"
  49.    semiJoinDone = false
  50.    systemFieldList = {RegularImmutableList@6230}  size = 0
  51.    condition = {RexCall@6231} "=($0, $2)"
  52.    variablesSet = {RegularImmutableSet@6232}  size = 0
  53.    joinType = {JoinRelType@6233} "LEFT"
  54.    joinInfo = {JoinInfo@6234} 
  55.    left = {RelSubset@6235} "rel#98:Subset#0.NONE.any.None: 0.false.UNKNOWN"
  56.    right = {RelSubset@6236} "rel#99:Subset#1.NONE.any.None: 0.false.UNKNOWN"
  57.    desc = "rel#100:LogicalJoin.NONE.any.None: 0.false.UNKNOWN(left=RelSubset#98,right=RelSubset#99,condition==($0, $2),joinType=left)"
  58.    rowType = {RelRecordType@6237} "RecordType(VARCHAR(2147483647) orderId, VARCHAR(2147483647) productName, VARCHAR(2147483647) orderId0, VARCHAR(2147483647) payType)"
  59.    digest = "LogicalJoin.NONE.any.None: 0.false.UNKNOWN(left=RelSubset#98,right=RelSubset#99,condition==($0, $2),joinType=left)"
  60.    cluster = {FlinkRelOptCluster@6239} 
  61.    id = 100
  62.    traitSet = {RelTraitSet@6240}  size = 5
  63.  planner = {VolcanoPlanner@6194} 
  64.  parents = null  
  65.   
  66. // 生成时候的调用栈   
  67.    
  68. create:106, FlinkLogicalJoin$ (org.apache.flink.table.planner.plan.nodes.logical)
  69. convert:92, FlinkLogicalJoinConverter (org.apache.flink.table.planner.plan.nodes.logical)
  70. onMatch:144, ConverterRule (org.apache.calcite.rel.convert)
  71. onMatch:208, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  72. findBestExp:631, VolcanoPlanner (org.apache.calcite.plan.volcano)
  73. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  74. optimize:64, FlinkVolcanoProgram (org.apache.flink.table.planner.plan.optimize.program)
  75. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  76. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  77. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  78. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  79. foreach:891, Iterator$class (scala.collection)
  80. foreach:1334, AbstractIterator (scala.collection)
  81. foreach:72, IterableLike$class (scala.collection)
  82. foreach:54, AbstractIterable (scala.collection)
  83. foldLeft:157, TraversableOnce$class (scala.collection)
  84. foldLeft:104, AbstractTraversable (scala.collection)
  85. optimize:57, FlinkChainedProgram (org.apache.flink.table.planner.plan.optimize.program)
  86. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  87. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  88. optimize:77, CommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  89. optimize:248, PlannerBase (org.apache.flink.table.planner.delegation)
  90. translate:151, PlannerBase (org.apache.flink.table.planner.delegation)
  91. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  92. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  93. toRetractStream:146, TableConversions (org.apache.flink.table.api.scala)
  94. main:75, SimpleOuterJoin$ (spendreport)
  95. main:-1, SimpleOuterJoin (spendreport)
  96.   
  97. abstract class FlinkLogicalJoinBase(  
  98.     cluster: RelOptCluster,
  99.     traitSet: RelTraitSet,
  100.     left: RelNode,
  101.     right: RelNode,
  102.     condition: RexNode,
  103.     joinType: JoinRelType)
  104.   extends Join(
  105.     cluster,
  106.     traitSet,
  107.     left,
  108.     right,
  109.     condition,
  110.     Set.empty[CorrelationId].asJava,
  111.     joinType)
  112.   with FlinkLogicalRel {
  113.     
  114.   // 这里也会计算cost
  115.   override def computeSelfCost(planner: RelOptPlanner, mq: RelMetadataQuery): RelOptCost = {
  116.     val leftRowCnt = mq.getRowCount(getLeft)
  117.     val leftRowSize = mq.getAverageRowSize(getLeft)
  118.     val rightRowCnt = mq.getRowCount(getRight)
  120.     joinType match {
  121.       case JoinRelType.SEMI | JoinRelType.ANTI =>
  122.         val rightRowSize = mq.getAverageRowSize(getRight)
  123.         val ioCost = (leftRowCnt * leftRowSize) + (rightRowCnt * rightRowSize)
  124.         val cpuCost = leftRowCnt + rightRowCnt
  125.         val rowCnt = leftRowCnt + rightRowCnt
  126.         planner.getCostFactory.makeCost(rowCnt, cpuCost, ioCost)
  127.       case _ =>
  128.         val cpuCost = leftRowCnt + rightRowCnt
  129.         val ioCost = (leftRowCnt * leftRowSize) + rightRowCnt
  130.         planner.getCostFactory.makeCost(leftRowCnt, cpuCost, ioCost)
  131.     }
  132.   }  
  133. }  
  134.   
  135. // 调用栈    
  136.   
  137. computeSelfCost:63, FlinkLogicalJoin (org.apache.flink.table.planner.plan.nodes.logical)
  138. getNonCumulativeCost:41, FlinkRelMdNonCumulativeCost (org.apache.flink.table.planner.plan.metadata)
  139. getNonCumulativeCost_$:-1, GeneratedMetadataHandler_NonCumulativeCost
  140. getNonCumulativeCost:-1, GeneratedMetadataHandler_NonCumulativeCost
  141. getNonCumulativeCost:301, RelMetadataQuery (org.apache.calcite.rel.metadata)
  142. getCost:936, VolcanoPlanner (org.apache.calcite.plan.volcano)
  143. propagateCostImprovements0:347, RelSubset (org.apache.calcite.plan.volcano)
  144. propagateCostImprovements:330, RelSubset (org.apache.calcite.plan.volcano)
  145. addRelToSet:1828, VolcanoPlanner (org.apache.calcite.plan.volcano)
  146. registerImpl:1764, VolcanoPlanner (org.apache.calcite.plan.volcano)
  147. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  148. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  149. ensureRegistered:1939, VolcanoPlanner (org.apache.calcite.plan.volcano)
  150. transformTo:129, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  151. transformTo:236, RelOptRuleCall (org.apache.calcite.plan)
  152. onMatch:146, ConverterRule (org.apache.calcite.rel.convert)
  153. onMatch:208, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  154. findBestExp:631, VolcanoPlanner (org.apache.calcite.plan.volcano)
  155. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  156. optimize:64, FlinkVolcanoProgram (org.apache.flink.table.planner.plan.optimize.program)
  157. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  158. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  159. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  160. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  161. foreach:891, Iterator$class (scala.collection)
  162. foreach:1334, AbstractIterator (scala.collection)
  163. foreach:72, IterableLike$class (scala.collection)
  164. foreach:54, AbstractIterable (scala.collection)
  165. foldLeft:157, TraversableOnce$class (scala.collection)
  166. foldLeft:104, AbstractTraversable (scala.collection)
  167. optimize:57, FlinkChainedProgram (org.apache.flink.table.planner.plan.optimize.program)
  168. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  169. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  170. optimize:77, CommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  171. optimize:248, PlannerBase (org.apache.flink.table.planner.delegation)
  172. translate:151, PlannerBase (org.apache.flink.table.planner.delegation)
  173. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  174. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  175. toRetractStream:146, TableConversions (org.apache.flink.table.api.scala)
  176. main:75, SimpleOuterJoin$ (spendreport)
  177. main:-1, SimpleOuterJoin (spendreport)

Calcite 会基于优化规则来优化这些 Logical Plan，根据运行环境的不同会应用不同的优化规则（Flink提供了批的优化规则 和 流的优化规则）。

优化规则分为两类，一类是Calcite提供的内置优化规则（如条件下推，剪枝等），另一类是是将Logical Node转变成 Flink Node 的规则。

这两步骤都属于 Calcite 的优化阶段。得到的 DataStream Plan 封装了如何将节点翻译成对应 DataStream / DataSet 程序的逻辑。其步骤就是将不同的 DataStream/DataSet Node 通过代码生成（CodeGen）翻译成最终可执行的 DataStream/DataSet 程序。

下面就列举了不同的Rule，每条规则会对应生成一个物理节点。比如节点内根据Calcite生成的sql的执行步骤，会进行codegen出DataSet的执行Function代码，

  1. package org.apache.Flink.table.plan.rules
  2.   
  3.   /**
  4.     * RuleSet to optimize plans for batch / DataSet execution
  5.     */
  6.   val DATASET_OPT_RULES: RuleSet = RuleSets.ofList(
  7.     // translate to Flink DataSet nodes
  8.     DataSetWindowAggregateRule.INSTANCE,
  9.     DataSetAggregateRule.INSTANCE,
  10.     DataSetDistinctRule.INSTANCE,
  11.     DataSetCalcRule.INSTANCE,
  12.     DataSetPythonCalcRule.INSTANCE,
  13.     DataSetJoinRule.INSTANCE,
  14.     DataSetSingleRowJoinRule.INSTANCE,
  15.     DataSetScanRule.INSTANCE,
  16.     DataSetUnionRule.INSTANCE,
  17.     DataSetIntersectRule.INSTANCE,
  18.     DataSetMinusRule.INSTANCE,
  19.     DataSetSortRule.INSTANCE,
  20.     DataSetValuesRule.INSTANCE,
  21.     DataSetCorrelateRule.INSTANCE,
  22.     BatchTableSourceScanRule.INSTANCE
  23.   )
  24.   
  25.    /**
  26.     * RuleSet to optimize plans for stream / DataStream execution
  27.     */
  28.   val DATASTREAM_OPT_RULES: RuleSet = RuleSets.ofList(
  29.     // translate to DataStream nodes
  30.     DataStreamSortRule.INSTANCE,
  31.     DataStreamGroupAggregateRule.INSTANCE,
  32.     DataStreamOverAggregateRule.INSTANCE,
  33.     DataStreamGroupWindowAggregateRule.INSTANCE,
  34.     DataStreamCalcRule.INSTANCE,
  35.     DataStreamScanRule.INSTANCE,
  36.     DataStreamUnionRule.INSTANCE,
  37.     DataStreamValuesRule.INSTANCE,
  38.     DataStreamCorrelateRule.INSTANCE,
  39.     DataStreamWindowJoinRule.INSTANCE,
  40.     DataStreamJoinRule.INSTANCE,
  41.     DataStreamTemporalTableJoinRule.INSTANCE,
  42.     StreamTableSourceScanRule.INSTANCE,
  43.     DataStreamMatchRule.INSTANCE,
  44.     DataStreamTableAggregateRule.INSTANCE,
  45.     DataStreamGroupWindowTableAggregateRule.INSTANCE,
  46.     DataStreamPythonCalcRule.INSTANCE
  47.   )
  48.     
  49. package org.apache.Flink.table.planner.plan.rules
  51.   /**
  52.     * RuleSet to do physical optimize for stream
  53.     */
  54.   val PHYSICAL_OPT_RULES: RuleSet = RuleSets.ofList(
  55.     FlinkExpandConversionRule.STREAM_INSTANCE,
  56.     // source
  57.     StreamExecDataStreamScanRule.INSTANCE,
  58.     StreamExecTableSourceScanRule.INSTANCE,
  59.     StreamExecIntermediateTableScanRule.INSTANCE,
  60.     StreamExecWatermarkAssignerRule.INSTANCE,
  61.     StreamExecValuesRule.INSTANCE,
  62.     // calc
  63.     StreamExecCalcRule.INSTANCE,
  64.     StreamExecPythonCalcRule.INSTANCE,
  65.     // union
  66.     StreamExecUnionRule.INSTANCE,
  67.     // sort
  68.     StreamExecSortRule.INSTANCE,
  69.     StreamExecLimitRule.INSTANCE,
  70.     StreamExecSortLimitRule.INSTANCE,
  71.     StreamExecTemporalSortRule.INSTANCE,
  72.     // rank
  73.     StreamExecRankRule.INSTANCE,
  74.     StreamExecDeduplicateRule.RANK_INSTANCE,
  75.     // expand
  76.     StreamExecExpandRule.INSTANCE,
  77.     // group agg
  78.     StreamExecGroupAggregateRule.INSTANCE,
  79.     StreamExecGroupTableAggregateRule.INSTANCE,
  80.     // over agg
  81.     StreamExecOverAggregateRule.INSTANCE,
  82.     // window agg
  83.     StreamExecGroupWindowAggregateRule.INSTANCE,
  84.     StreamExecGroupWindowTableAggregateRule.INSTANCE,
  85.     // join
  86.     StreamExecJoinRule.INSTANCE,
  87.     StreamExecWindowJoinRule.INSTANCE,
  88.     StreamExecTemporalJoinRule.INSTANCE,
  89.     StreamExecLookupJoinRule.SNAPSHOT_ON_TABLESCAN,
  90.     StreamExecLookupJoinRule.SNAPSHOT_ON_CALC_TABLESCAN,
  91.     // CEP
  92.     StreamExecMatchRule.INSTANCE,
  93.     // correlate
  94.     StreamExecConstantTableFunctionScanRule.INSTANCE,
  95.     StreamExecCorrelateRule.INSTANCE,
  96.     // sink
  97.     StreamExecSinkRule.INSTANCE
  98.   )

一个具体的Rule举例 ，这里是 Union 的 Rule ：

  1. package org.apache.Flink.table.planner.plan.rules.physical.stream
  3. class StreamExecUnionRule
  4.   extends ConverterRule(
  5.     classOf[FlinkLogicalUnion],
  6.     FlinkConventions.LOGICAL,
  7.     FlinkConventions.STREAM_PHYSICAL,
  8.     "StreamExecUnionRule") {
  9.   
  10.   def convert(rel: RelNode): RelNode = {
  11.     val union: FlinkLogicalUnion = rel.asInstanceOf[FlinkLogicalUnion]
  12.     val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConventions.STREAM_PHYSICAL)
  13.     val newInputs = union.getInputs.map(RelOptRule.convert(_, FlinkConventions.STREAM_PHYSICAL))
  15.     // 这里本条规则会对应生成一个物理节点。节点内根据Calcite生成的sql的执行步骤，会进行codegen出Stream的执行Function代码，
  16.     new StreamExecUnion(
  17.       rel.getCluster,
  18.       traitSet,
  19.       newInputs,
  20.       union.all,
  21.       rel.getRowType)
  22.     }
  23.   }  
  24. } 
  26. public class VolcanoPlanner extends AbstractRelOptPlanner {
  27.     public RelNode findBestExp() {
  28.              // 在这里会对Rule进行匹配调用
  29.              match.onMatch();
  30.         return cheapest;
  31.     }
  32. }
  34. match = {VolcanoRuleMatch@6252} "rule [StreamExecUnionRule(in:LOGICAL,out:STREAM_PHYSICAL)] rels [rel#215:FlinkLogicalUnion.LOGICAL.any.None: 0.false.UNKNOWN(input#0=RelSubset#211,input#1=RelSubset#214,all=true)]"
  35.  targetSet = {RelSet@6298} 
  36.  targetSubset = null
  37.  digest = "rule [StreamExecUnionRule(in:LOGICAL,out:STREAM_PHYSICAL)] rels [rel#215:FlinkLogicalUnion.LOGICAL.any.None: 0.false.UNKNOWN(input#0=RelSubset#211,input#1=RelSubset#214,all=true)]"
  38.  cachedImportance = 0.81
  39.  volcanoPlanner = {VolcanoPlanner@6259} 
  40.  generatedRelList = null
  41.  id = 521
  42.  operand0 = {RelOptRule$ConverterRelOptRuleOperand@6247} 
  43.  nodeInputs = {RegularImmutableBiMap@6299}  size = 0
  44.  rule = {StreamExecUnionRule@6241} "StreamExecUnionRule(in:LOGICAL,out:STREAM_PHYSICAL)"
  45.  rels = {RelNode[1]@6300} 
  46.  planner = {VolcanoPlanner@6259} 
  47.  parents = null
  49. // 调用栈       
  50.    
  51. create:106, FlinkLogicalJoin$ (org.apache.flink.table.planner.plan.nodes.logical)
  52. convert:92, FlinkLogicalJoinConverter (org.apache.flink.table.planner.plan.nodes.logical)
  53. onMatch:144, ConverterRule (org.apache.calcite.rel.convert)
  54. onMatch:208, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  55. findBestExp:631, VolcanoPlanner (org.apache.calcite.plan.volcano)
  56. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  57. optimize:64, FlinkVolcanoProgram (org.apache.flink.table.planner.plan.optimize.program)
  58. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  59. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  60. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  61. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  62. foreach:891, Iterator$class (scala.collection)
  63. foreach:1334, AbstractIterator (scala.collection)
  64. foreach:72, IterableLike$class (scala.collection)
  65. foreach:54, AbstractIterable (scala.collection)
  66. foldLeft:157, TraversableOnce$class (scala.collection)
  67. foldLeft:104, AbstractTraversable (scala.collection)
  68. optimize:57, FlinkChainedProgram (org.apache.flink.table.planner.plan.optimize.program)
  69. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  70. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  71. optimize:77, CommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  72. optimize:248, PlannerBase (org.apache.flink.table.planner.delegation)
  73. translate:151, PlannerBase (org.apache.flink.table.planner.delegation)
  74. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  75. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  76. toRetractStream:146, TableConversions (org.apache.flink.table.api.scala)
  77. main:75, SimpleOuterJoin$ (spendreport)
  78. main:-1, SimpleOuterJoin (spendreport)
  79.    
  80.   
  81. // 调用栈      
  82.   
  83. convert:46, StreamExecUnionRule (org.apache.Flink.table.planner.plan.rules.physical.stream)
  84. onMatch:144, ConverterRule (org.apache.calcite.rel.convert)
  85. onMatch:208, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  86. findBestExp:631, VolcanoPlanner (org.apache.calcite.plan.volcano)
  87. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  88. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  89. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  90. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  91. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  92. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  93. foreach:891, Iterator$class (scala.collection)
  94. foreach:1334, AbstractIterator (scala.collection)
  95. foreach:72, IterableLike$class (scala.collection)
  96. foreach:54, AbstractIterable (scala.collection)
  97. foldLeft:157, TraversableOnce$class (scala.collection)
  98. foldLeft:104, AbstractTraversable (scala.collection)
  99. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  100. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  101. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  102. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  103. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  104. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  105. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  106. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  107. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  108. main:89, StreamSQLExample$ (spendreport)
  109. main:-1, StreamSQLExample (spendreport)

另一个具体的Rule举例 ，这里是 Join的优化，StreamExecJoin的生成

  1. class StreamExecJoinRule {
  2.     override def onMatch(call: RelOptRuleCall): Unit = {
  3.       val newJoin = new StreamExecJoin(
  4.         join.getCluster,
  5.         providedTraitSet,
  6.         newLeft,
  7.         newRight,
  8.         join.getCondition,
  9.         join.getJoinType)
  10.       call.transformTo(newJoin)
  11.     }
  12. }
  14. newJoin = {StreamExecJoin@6326} "StreamExecJoin#152"
  15.  cluster = {FlinkRelOptCluster@5072} 
  16.  joinType = {JoinRelType@5038} "LEFT"
  17.  LOG = null
  18.  transformation = null
  19.  bitmap$trans$0 = false
  20.  CommonPhysicalJoin.joinType = {JoinRelType@5038} "LEFT"
  21.  filterNulls = null
  22.  keyPairs = null
  23.  flinkJoinType = null
  24.  inputRowType = null
  25.  bitmap$0 = 0
  26.  condition = {RexCall@5041} "=($0, $2)"
  27.  variablesSet = {RegularImmutableSet@6342}  size = 0
  28.  Join.joinType = {JoinRelType@5038} "LEFT"
  29.  joinInfo = {JoinInfo@6343} 
  30.  left = {RelSubset@6328} "rel#150:Subset#5.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  31.   bestCost = {FlinkCost$$anon$1@6344} "{inf}"
  32.   set = {RelSet@6348} 
  33.   best = null
  34.   timestamp = 0
  35.   boosted = false
  36.   desc = "rel#150:Subset#5.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  37.   rowType = {RelRecordType@6349} "RecordType(VARCHAR(2147483647) orderId, VARCHAR(2147483647) productName)"
  38.   digest = "Subset#5.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  39.   cluster = {FlinkRelOptCluster@5072} 
  40.   id = 150
  41.   traitSet = {RelTraitSet@6336}  size = 5
  42.  right = {RelSubset@6329} "rel#151:Subset#6.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  43.   bestCost = {FlinkCost$$anon$1@6344} "{inf}"
  44.   set = {RelSet@6345} 
  45.   best = null
  46.   timestamp = 0
  47.   boosted = false
  48.   desc = "rel#151:Subset#6.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  49.   rowType = null
  50.   digest = "Subset#6.STREAM_PHYSICAL.hash[0]true.None: 0.false.UNKNOWN"
  51.   cluster = {FlinkRelOptCluster@5072} 
  52.   id = 151
  53.   traitSet = {RelTraitSet@6336}  size = 5
  54.  desc = "StreamExecJoin#152"
  55.  rowType = null
  56.  digest = "StreamExecJoin#152"
  57.  AbstractRelNode.cluster = {FlinkRelOptCluster@5072} 
  58.  id = 152
  59.  traitSet = {RelTraitSet@6327}  size = 5
  61. // 调用栈       
  62.    
  63. <init>:58, StreamExecJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  64. onMatch:128, StreamExecJoinRule (org.apache.flink.table.planner.plan.rules.physical.stream)
  65. onMatch:208, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  66. findBestExp:631, VolcanoPlanner (org.apache.calcite.plan.volcano)
  67. run:327, Programs$RuleSetProgram (org.apache.calcite.tools)
  68. optimize:64, FlinkVolcanoProgram (org.apache.flink.table.planner.plan.optimize.program)
  69. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  70. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  71. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  72. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  73. foreach:891, Iterator$class (scala.collection)
  74. foreach:1334, AbstractIterator (scala.collection)
  75. foreach:72, IterableLike$class (scala.collection)
  76. foreach:54, AbstractIterable (scala.collection)
  77. foldLeft:157, TraversableOnce$class (scala.collection)
  78. foldLeft:104, AbstractTraversable (scala.collection)
  79. optimize:57, FlinkChainedProgram (org.apache.flink.table.planner.plan.optimize.program)
  80. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  81. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  82. optimize:77, CommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  83. optimize:248, PlannerBase (org.apache.flink.table.planner.delegation)
  84. translate:151, PlannerBase (org.apache.flink.table.planner.delegation)
  85. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  86. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  87. toRetractStream:146, TableConversions (org.apache.flink.table.api.scala)
  88. main:75, SimpleOuterJoin$ (spendreport)
  89. main:-1, SimpleOuterJoin (spendreport)

这时候脉络图如下

  1. // NOTE : 执行顺序是从上至下，" -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |      
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |     
  12. *     SqlToRelConverter.convertQuery // 语义分析，生成逻辑计划，作用是SqlNode–>RelNode
  13. *        |    
  14. *        |     
  15. *        +-----> LogicalProject (RelNode) // Abstract Syntax Tree，未优化的RelNode   
  16. *        |      
  17. *        |     
  18. *    FlinkLogicalJoinConverter (RelOptRule) // Flink定制的优化rules      
  19. *    VolcanoRuleCall.onMatch // 基于Flink定制的一些优化rules去优化 Logical Plan 
  20. *        | 
  21. *        |   
  22. *        +-----> FlinkLogicalJoin (RelNode)  // Optimized Logical Plan，逻辑执行计划
  23. *        |  
  24. *        |    
  25. *    StreamExecJoinRule (RelOptRule) // Rule that converts FlinkLogicalJoin without window bounds in join condition to StreamExecJoin     
  26. *    VolcanoRuleCall.onMatch // 基于Flink rules将optimized LogicalPlan转成Flink物理执行计划
  27. *        |       
  28. *        |   
  29. *        +-----> StreamExecJoin (FlinkRelNode) // Stream physical RelNode，物理执行计划
  30. *        |      
  31. *        |   
  32. *    StreamExecJoin.translateToPlanInternal  // 作用是生成 StreamOperator, 即Flink算子  
  33. *        |     
  34. *        |     
  35. *        +-----> StreamingJoinOperator (StreamOperator) // Streaming unbounded Join operator in StreamTask 
  36. *        |     
  37. *        |

Calcite 针对不同的大数据组件，将优化后的plan映射到最终的大数据引擎，如折射成Flink图。

这一块只要是递归调用各个节点DataStreamRel的translateToPlan方法，这个方法利用CodeGen元编程成Flink的各种算子。现在就相当于我们直接利用Flink的DataSet或DataStream API开发的程序。

  1. class StreamPlanner(
  2.     executor: Executor,
  3.     config: TableConfig,
  4.     functionCatalog: FunctionCatalog,
  5.     catalogManager: CatalogManager)
  6.   extends PlannerBase(executor, config, functionCatalog, catalogManager, isStreamingMode = true) {
  7.   override protected def translateToPlan(
  8.       execNodes: util.List[ExecNode[_, _]]): util.List[Transformation[_]] = {
  9.     execNodes.map {
  10.       case node: StreamExecNode[_] => node.translateToPlan(this)
  11.       case _ =>
  12.         throw new TableException("Cannot generate DataStream due to an invalid logical plan. " +
  13.           "This is a bug and should not happen. Please file an issue.")
  14.     }
  15.   }
  16. }
  18. package org.apache.Flink.table.planner.plan.nodes.physical.stream	
  20. class StreamExecUnion(
  21.     cluster: RelOptCluster,
  22.     traitSet: RelTraitSet,
  23.     inputRels: util.List[RelNode],
  24.     all: Boolean,
  25.     outputRowType: RelDataType)
  26.   extends Union(cluster, traitSet, inputRels, all)
  27.   with StreamPhysicalRel
  28.   with StreamExecNode[BaseRow] {
  30.   // 这里就生成了Flink算子
  31.   override protected def translateToPlanInternal(
  32.       planner: StreamPlanner): Transformation[BaseRow] = {
  33.     val transformations = getInputNodes.map {
  34.       input => input.translateToPlan(planner).asInstanceOf[Transformation[BaseRow]]
  35.     }
  36.     new UnionTransformation(transformations)
  37.   }
  38. }
  40.  // 调用栈
  42. translateToPlanInternal:85, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  43. translateToPlanInternal:39, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  44. translateToPlan:58, ExecNode$class (org.apache.Flink.table.planner.plan.nodes.exec)
  45. translateToPlan:39, StreamExecUnion (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  46. translateToTransformation:184, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  47. translateToPlanInternal:153, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  48. translateToPlanInternal:48, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  49. translateToPlan:58, ExecNode$class (org.apache.Flink.table.planner.plan.nodes.exec)
  50. translateToPlan:48, StreamExecSink (org.apache.Flink.table.planner.plan.nodes.physical.stream)
  51. apply:60, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.Flink.table.planner.delegation)
  52. apply:59, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.Flink.table.planner.delegation)
  53. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  54. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  55. foreach:891, Iterator$class (scala.collection)
  56. foreach:1334, AbstractIterator (scala.collection)
  57. foreach:72, IterableLike$class (scala.collection)
  58. foreach:54, AbstractIterable (scala.collection)
  59. map:234, TraversableLike$class (scala.collection)
  60. map:104, AbstractTraversable (scala.collection)
  61. translateToPlan:59, StreamPlanner (org.apache.Flink.table.planner.delegation)
  62. translate:153, PlannerBase (org.apache.Flink.table.planner.delegation)
  63. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  64. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  65. toAppendStream:101, TableConversions (org.apache.Flink.table.api.scala)
  66. main:89, StreamSQLExample$ (spendreport)
  67. main:-1, StreamSQLExample (spendreport)

此时脉络图补充完全。

  1. // NOTE : 执行顺序是从上至下，" -----> " 表示生成的实例类型
  2. * 
  3. *        +-----> "left outer JOIN" (SQL statement)
  4. *        |   
  5. *        |     
  6. *     SqlParser.parseQuery // SQL 解析阶段，生成AST（抽象语法树），作用是SQL–>SqlNode      
  7. *        |   
  8. *        |      
  9. *        +-----> SqlJoin (SqlNode)
  10. *        |   
  11. *        |     
  12. *     SqlToRelConverter.convertQuery // 语义分析，生成逻辑计划，作用是SqlNode–>RelNode
  13. *        |    
  14. *        |     
  15. *        +-----> LogicalProject (RelNode) // Abstract Syntax Tree，未优化的RelNode   
  16. *        |      
  17. *        |     
  18. *    FlinkLogicalJoinConverter (RelOptRule) // Flink定制的优化rules      
  19. *    VolcanoRuleCall.onMatch // 基于Flink定制的一些优化rules去优化 Logical Plan 
  20. *        | 
  21. *        |   
  22. *        +-----> FlinkLogicalJoin (RelNode)  // Optimized Logical Plan，逻辑执行计划
  23. *        |  
  24. *        |    
  25. *    StreamExecJoinRule (RelOptRule) // Rule that converts FlinkLogicalJoin without window bounds in join condition to StreamExecJoin     
  26. *    VolcanoRuleCall.onMatch // 基于Flink rules将optimized LogicalPlan转成Flink物理执行计划
  27. *        |       
  28. *        |   
  29. *        +-----> StreamExecJoin (FlinkRelNode) // Stream physical RelNode，物理执行计划
  30. *        |      
  31. *        |     
  32. *    StreamExecJoin.translateToPlanInternal  // 作用是生成 StreamOperator, 即Flink算子  
  33. *        |     
  34. *        |     
  35. *        +-----> StreamingJoinOperator (StreamOperator) // Streaming unbounded Join operator in StreamTask 
  36. *        |     
  37. *        |       
  38. *    StreamTwoInputProcessor.processRecord1// 在TwoInputStreamTask调用StreamingJoinOperator，真实的执行  
  39. *        |
  40. *        |

运行时候，则会在StreamTask中进行业务操作，这就是我们熟悉的操作了。调用栈举例如下

  1. processElement:150, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  2. emitNext:128, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  3. processInput:69, StreamOneInputProcessor (org.apache.Flink.streaming.runtime.io)
  4. processInput:311, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  5. runDefaultAction:-1, 354713989 (org.apache.Flink.streaming.runtime.tasks.StreamTask$$Lambda$710)
  6. runMailboxLoop:187, MailboxProcessor (org.apache.Flink.streaming.runtime.tasks.mailbox)
  7. runMailboxLoop:487, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  8. invoke:470, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  9. doRun:707, Task (org.apache.Flink.runtime.taskmanager)
  10. run:532, Task (org.apache.Flink.runtime.taskmanager)
  11. run:748, Thread (java.lang)

下面是如何具体生成各种执行计划的代码

  1. import org.apache.Flink.api.java.utils.ParameterTool
  2. import org.apache.Flink.api.scala._
  3. import org.apache.Flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
  4. import org.apache.Flink.table.api.EnvironmentSettings
  5. import org.apache.Flink.table.api.scala._
  7. object StreamSQLExample {
  9.   // *************************************************************************
  10.   //     PROGRAM
  11.   // *************************************************************************
  12.   def main(args: Array[String]): Unit = {
  14.     val params = ParameterTool.fromArgs(args)
  15.     val planner = if (params.has("planner")) params.get("planner") else "Flink"
  17.     // set up execution environment
  18.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  19.     val tEnv = if (planner == "blink") {  // use blink planner in streaming mode
  20.       val settings = EnvironmentSettings.newInstance()
  21.         .useBlinkPlanner()
  22.         .inStreamingMode()
  23.         .build()
  24.       StreamTableEnvironment.create(env, settings)
  25.     } else if (planner == "Flink") {  // use Flink planner in streaming mode
  26.       StreamTableEnvironment.create(env)
  27.     } else {
  28.       System.err.println("The planner is incorrect. Please run 'StreamSQLExample --planner <planner>', " +
  29.         "where planner (it is either Flink or blink, and the default is Flink) indicates whether the " +
  30.         "example uses Flink planner or blink planner.")
  31.       return
  32.     }
  34.     val orderA: DataStream[Order] = env.fromCollection(Seq(
  35.       Order(1L, "beer", 3),
  36.       Order(1L, "diaper", 4),
  37.       Order(3L, "rubber", 2)))
  39.     val orderB: DataStream[Order] = env.fromCollection(Seq(
  40.       Order(2L, "pen", 3),
  41.       Order(2L, "rubber", 3),
  42.       Order(4L, "beer", 1)))
  44.     // convert DataStream to Table
  45.     val tableA = tEnv.fromDataStream(orderA, 'user, 'product, 'amount)
  46.     // register DataStream as Table
  47.     tEnv.registerDataStream("OrderB", orderB, 'user, 'product, 'amount)
  49.     // union the two tables
  50.     val result = tEnv.sqlQuery(
  51.       s"""
  52.          |SELECT * FROM $tableA WHERE amount > 2
  53.          |UNION ALL
  54.          |SELECT * FROM OrderB WHERE amount < 2
  55.         """.stripMargin)
  57.     result.toAppendStream[Order].print()
  58.     print(tEnv.explain(result))
  59.     env.execute()
  60.   }
  62.   // *************************************************************************
  63.   //     USER DATA TYPES
  64.   // *************************************************************************
  65.   case class Order(user: Long, product: String, amount: Int)
  66. }

整个流程的转换大体就像这样：

  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3. :- LogicalProject(user=[$0], product=[$1], amount=[$2])
  4. :  +- LogicalFilter(condition=[>($2, 2)])
  5. :     +- LogicalTableScan(table=[[default_catalog, default_database, UnnamedTable$0]])
  6. +- LogicalProject(user=[$0], product=[$1], amount=[$2])
  7.    +- LogicalFilter(condition=[<($2, 2)])
  8.       +- LogicalTableScan(table=[[default_catalog, default_database, OrderB]])
  10. == Optimized Logical Plan ==
  11. Union(all=[true], union=[user, product, amount])
  12. :- Calc(select=[user, product, amount], where=[>(amount, 2)])
  13. :  +- DataStreamScan(table=[[default_catalog, default_database, UnnamedTable$0]], fields=[user, product, amount])
  14. +- Calc(select=[user, product, amount], where=[<(amount, 2)])
  15.    +- DataStreamScan(table=[[default_catalog, default_database, OrderB]], fields=[user, product, amount])
  17. == Physical Execution Plan ==
  18. Stage 1 : Data Source
  19. 	content : Source: Collection Source
  21. Stage 2 : Data Source
  22. 	content : Source: Collection Source
  24. 	Stage 10 : Operator
  25. 		content : SourceConversion(table=[default_catalog.default_database.UnnamedTable$0], fields=[user, product, amount])
  26. 		ship_strategy : FORWARD
  28. 		Stage 11 : Operator
  29. 			content : Calc(select=[user, product, amount], where=[(amount > 2)])
  30. 			ship_strategy : FORWARD
  32. 			Stage 12 : Operator
  33. 				content : SourceConversion(table=[default_catalog.default_database.OrderB], fields=[user, product, amount])
  34. 				ship_strategy : FORWARD
  36. 				Stage 13 : Operator
  37. 					content : Calc(select=[user, product, amount], where=[(amount < 2)])
  38. 					ship_strategy : FORWARD
  1. import java.sql.Timestamp
  2. import org.apache.Flink.api.java.utils.ParameterTool
  3. import org.apache.Flink.api.scala._
  4. import org.apache.Flink.streaming.api.TimeCharacteristic
  5. import org.apache.Flink.streaming.api.scala.StreamExecutionEnvironment
  6. import org.apache.Flink.table.api.{EnvironmentSettings, TableEnvironment}
  7. import org.apache.Flink.table.api.scala._
  8. import org.apache.Flink.types.Row
  10. import scala.collection.mutable
  12. object SimpleOuterJoin {
  13.   def main(args: Array[String]): Unit = {
  15.     val params = ParameterTool.fromArgs(args)
  16.     val planner = if (params.has("planner")) params.get("planner") else "Flink"
  18.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  20.     val tEnv = if (planner == "blink") {  // use blink planner in streaming mode
  21.       val settings = EnvironmentSettings.newInstance()
  22.         .useBlinkPlanner()
  23.         .inStreamingMode()
  24.         .build()
  25.       StreamTableEnvironment.create(env, settings)
  26.     } else if (planner == "Flink") {  // use Flink planner in streaming mode
  27.       StreamTableEnvironment.create(env)
  28.     } else {
  29.       System.err.println("The planner is incorrect. Please run 'StreamSQLExample --planner <planner>', " +
  30.         "where planner (it is either Flink or blink, and the default is Flink) indicates whether the " +
  31.         "example uses Flink planner or blink planner.")
  32.       return
  33.     }
  35.     env.setParallelism(1)
  36.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  37.     // 构造订单数据
  38.     val ordersData = new mutable.MutableList[(String, String)]
  39.     ordersData.+=(("001", "iphone"))
  40.     ordersData.+=(("002", "mac"))
  41.     ordersData.+=(("003", "book"))
  42.     ordersData.+=(("004", "cup"))
  44.     // 构造付款表
  45.     val paymentData = new mutable.MutableList[(String, String)]
  46.     paymentData.+=(("001", "alipay"))
  47.     paymentData.+=(("002", "card"))
  48.     paymentData.+=(("003", "card"))
  49.     paymentData.+=(("004", "alipay"))
  50.     val orders = env
  51.       .fromCollection(ordersData)
  52.        .toTable(tEnv, 'orderId, 'productName)
  53.     val ratesHistory = env
  54.       .fromCollection(paymentData)
  55.       .toTable(tEnv, 'orderId, 'payType)
  57.     tEnv.registerTable("Orders", orders)
  58.     tEnv.registerTable("Payment", ratesHistory)
  60.     var sqlQuery =
  61.       """
  62.         |SELECT
  63.         |  o.orderId,
  64.         |  o.productName,
  65.         |  p.payType
  66.         |FROM
  67.         |  Orders AS o left outer JOIN Payment AS p ON o.orderId = p.orderId
  68.         |""".stripMargin
  69.     tEnv.registerTable("TemporalJoinResult", tEnv.sqlQuery(sqlQuery))
  71.     val result = tEnv.scan("TemporalJoinResult").toRetractStream[Row]
  72.     result.print()
  73.     print(tEnv.explain(tEnv.sqlQuery(sqlQuery)))
  74.     env.execute()
  75.   }
  76. }

整个流程的转换如下：

  1. == Abstract Syntax Tree ==
  2. LogicalProject(orderId=[$0], productName=[$1], payType=[$3])
  3. +- LogicalJoin(condition=[=($0, $2)], joinType=[left])
  4.    :- LogicalTableScan(table=[[default_catalog, default_database, Orders]])
  5.    +- LogicalTableScan(table=[[default_catalog, default_database, Payment]])
  7. == Optimized Logical Plan ==
  8. Calc(select=[orderId, productName, payType])
  9. +- Join(joinType=[LeftOuterJoin], where=[=(orderId, orderId0)], select=[orderId, productName, orderId0, payType], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
  10.    :- Exchange(distribution=[hash[orderId]])
  11.    :  +- DataStreamScan(table=[[default_catalog, default_database, Orders]], fields=[orderId, productName])
  12.    +- Exchange(distribution=[hash[orderId]])
  13.       +- DataStreamScan(table=[[default_catalog, default_database, Payment]], fields=[orderId, payType])
  15. == Physical Execution Plan ==
  16. Stage 1 : Data Source
  17. 	content : Source: Collection Source
  19. Stage 2 : Data Source
  20. 	content : Source: Collection Source
  22. 	Stage 11 : Operator
  23. 		content : SourceConversion(table=[default_catalog.default_database.Orders], fields=[orderId, productName])
  24. 		ship_strategy : FORWARD
  26. 		Stage 13 : Operator
  27. 			content : SourceConversion(table=[default_catalog.default_database.Payment], fields=[orderId, payType])
  28. 			ship_strategy : FORWARD
  30. 			Stage 15 : Operator
  31. 				content : Join(joinType=[LeftOuterJoin], where=[(orderId = orderId0)], select=[orderId, productName, orderId0, payType], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
  32. 				ship_strategy : HASH
  34. 				Stage 16 : Operator
  35. 					content : Calc(select=[orderId, productName, payType])
  36. 					ship_strategy : FORWARD
  38. 输出结果是
  39. (true,001,iphone,null)
  40. (false,001,iphone,null)
  41. (true,001,iphone,alipay)
  42. (true,002,mac,null)
  43. (false,002,mac,null)
  44. (true,002,mac,card)
  45. (true,003,book,null)
  46. (false,003,book,null)
  47. (true,003,book,card)
  48. (true,004,cup,null)
  49. (false,004,cup,null)
  50. (true,004,cup,alipay)

下面是调试时候的调用栈，这个可以给大家参考

  1. // 调用Rule进行优化
  3. matches:49, StreamExecJoinRule (org.apache.Flink.table.planner.plan.rules.physical.stream)
  4. matchRecurse:263, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  5. matchRecurse:370, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  6. matchRecurse:370, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  7. match:247, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  8. fireRules:1534, VolcanoPlanner (org.apache.calcite.plan.volcano)
  9. registerImpl:1807, VolcanoPlanner (org.apache.calcite.plan.volcano)
  10. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  11. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  12. ensureRegistered:90, VolcanoPlanner (org.apache.calcite.plan.volcano)
  13. onRegister:329, AbstractRelNode (org.apache.calcite.rel)
  14. registerImpl:1668, VolcanoPlanner (org.apache.calcite.plan.volcano)
  15. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  16. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  17. ensureRegistered:90, VolcanoPlanner (org.apache.calcite.plan.volcano)
  18. onRegister:329, AbstractRelNode (org.apache.calcite.rel)
  19. registerImpl:1668, VolcanoPlanner (org.apache.calcite.plan.volcano)
  20. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  21. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  22. changeTraits:529, VolcanoPlanner (org.apache.calcite.plan.volcano)
  23. run:324, Programs$RuleSetProgram (org.apache.calcite.tools)
  24. optimize:64, FlinkVolcanoProgram (org.apache.Flink.table.planner.plan.optimize.program)
  25. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  26. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.Flink.table.planner.plan.optimize.program)
  27. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  28. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  29. foreach:891, Iterator$class (scala.collection)
  30. foreach:1334, AbstractIterator (scala.collection)
  31. foreach:72, IterableLike$class (scala.collection)
  32. foreach:54, AbstractIterable (scala.collection)
  33. foldLeft:157, TraversableOnce$class (scala.collection)
  34. foldLeft:104, AbstractTraversable (scala.collection)
  35. optimize:57, FlinkChainedProgram (org.apache.Flink.table.planner.plan.optimize.program)
  36. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  37. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  38. optimize:77, CommonSubGraphBasedOptimizer (org.apache.Flink.table.planner.plan.optimize)
  39. optimize:248, PlannerBase (org.apache.Flink.table.planner.delegation)
  40. translate:151, PlannerBase (org.apache.Flink.table.planner.delegation)
  41. toDataStream:210, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  42. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.Flink.table.api.scala.internal)
  43. toRetractStream:146, TableConversions (org.apache.Flink.table.api.scala)
  44. main:75, SimpleOuterJoin$ (spendreport)
  45. main:-1, SimpleOuterJoin (spendreport)
  46.   
  47.   
  48. // 调用Rule进行转换到Flink逻辑算子  
  49.   
  50. translateToPlanInternal:140, StreamExecJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  51. translateToPlanInternal:51, StreamExecJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  52. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  53. translateToPlan:51, StreamExecJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  54. translateToPlanInternal:54, StreamExecCalc (org.apache.flink.table.planner.plan.nodes.physical.stream)
  55. translateToPlanInternal:39, StreamExecCalc (org.apache.flink.table.planner.plan.nodes.physical.stream)
  56. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  57. translateToPlan:38, StreamExecCalcBase (org.apache.flink.table.planner.plan.nodes.physical.stream)
  58. translateToTransformation:184, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  59. translateToPlanInternal:153, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  60. translateToPlanInternal:48, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  61. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  62. translateToPlan:48, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  63. apply:60, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.flink.table.planner.delegation)
  64. apply:59, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.flink.table.planner.delegation)
  65. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  66. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  67. foreach:891, Iterator$class (scala.collection)
  68. foreach:1334, AbstractIterator (scala.collection)
  69. foreach:72, IterableLike$class (scala.collection)
  70. foreach:54, AbstractIterable (scala.collection)
  71. map:234, TraversableLike$class (scala.collection)
  72. map:104, AbstractTraversable (scala.collection)
  73. translateToPlan:59, StreamPlanner (org.apache.flink.table.planner.delegation)
  74. translate:153, PlannerBase (org.apache.flink.table.planner.delegation)
  75. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  76. toRetractStream:127, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  77. toRetractStream:146, TableConversions (org.apache.flink.table.api.scala)
  78. main:75, SimpleOuterJoin$ (spendreport)
  79. main:-1, SimpleOuterJoin (spendreport)  
  80.  
  81. // 运行时候
  82.   
  83. @Internal
  84. public final class StreamTwoInputProcessor<IN1, IN2> implements StreamInputProcessor {
  85. 	private void processRecord2(
  86. 			StreamRecord<IN2> record,
  87. 			TwoInputStreamOperator<IN1, IN2, ?> streamOperator,
  88. 			Counter numRecordsIn) throws Exception {
  90. 		streamOperator.setKeyContextElement2(record);
  91. 		streamOperator.processElement2(record);
  92. 		postProcessRecord(numRecordsIn);
  93. 	}  
  94. }    
  96. // 能看出来，streamOperator就是StreamingJoinOperator
  98. streamOperator = {StreamingJoinOperator@10943} 
  99.  leftIsOuter = true
  100.  rightIsOuter = false
  101.  outRow = {JoinedRow@10948} "JoinedRow{row1=org.apache.flink.table.dataformat.BinaryRow@dc6a1b67, row2=(+|null,null)}"
  102.  leftNullRow = {GenericRow@10949} "(+|null,null)"
  103.  rightNullRow = {GenericRow@10950} "(+|null,null)"
  104.  leftRecordStateView = {OuterJoinRecordStateViews$InputSideHasNoUniqueKey@10945} 
  105.  rightRecordStateView = {JoinRecordStateViews$InputSideHasNoUniqueKey@10946} 
  106.  generatedJoinCondition = {GeneratedJoinCondition@10951} 
  107.  leftType = {BaseRowTypeInfo@10952} "BaseRow(orderId: STRING, productName: STRING)"
  108.  rightType = {BaseRowTypeInfo@10953} "BaseRow(orderId: STRING, payType: STRING)"
  109.  leftInputSideSpec = {JoinInputSideSpec@10954} "NoUniqueKey"
  110.  rightInputSideSpec = {JoinInputSideSpec@10955} "NoUniqueKey"
  111.  nullFilterKeys = {int[1]@10956} 
  112.  nullSafe = false
  113.  filterAllNulls = true
  114.  minRetentionTime = 0
  115.  stateCleaningEnabled = false
  116.  joinCondition = {AbstractStreamingJoinOperator$JoinConditionWithNullFilters@10947} 
  117.  collector = {TimestampedCollector@10957} 
  118.  chainingStrategy = {ChainingStrategy@10958} "HEAD"
  119.  container = {TwoInputStreamTask@10959} "Join(joinType=[LeftOuterJoin], where=[(orderId = orderId0)], select=[orderId, productName, orderId0, payType], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey]) -> Calc(select=[orderId, productName, payType]) -> SinkConversionToTuple2 -> Sink: Print to Std. Out (1/1)"
  120.  config = {StreamConfig@10960} "\n=======================Stream Config=======================\nNumber of non-chained inputs: 2\nNumber of non-chained outputs: 0\nOutput names: []\nPartitioning:\nChained subtasks: [(Join(joinType=[LeftOuterJoin], where=[(orderId = orderId0)], select=[orderId, productName, orderId0, payType], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])-7 -> Calc(select=[orderId, productName, payType])-8, typeNumber=0, selectedNames=[], outputPartitioner=FORWARD, outputTag=null)]\nOperator: SimpleOperatorFactory\nBuffer timeout: 100\nState Monitoring: false\n\n\n---------------------\nChained task configs\n---------------------\n{8=\n=======================Stream Config=======================\nNumber of non-chained inputs: 0\nNumber of non-chained outputs: 0\nOutput names: []\nPartitioning:\nChained subtasks: [(Calc(select=[orderId, productName, payType])-8 -> SinkConversionToTuple2-9, typeNumber=0, selectedNames=[], outputPartitioner=FORWARD, outputTag=null)]\nOperator: CodeGenOperatorFactory\nBuffer timeout: "
  121.  output = {AbstractStreamOperator$CountingOutput@10961} 
  122.  runtimeContext = {StreamingRuntimeContext@10962} 
  123.  stateKeySelector1 = {BinaryRowKeySelector@10963} 
  124.  stateKeySelector2 = {BinaryRowKeySelector@10964} 
  125.  keyedStateBackend = {HeapKeyedStateBackend@10965} "HeapKeyedStateBackend"
  126.  keyedStateStore = {DefaultKeyedStateStore@10966} 
  127.  operatorStateBackend = {DefaultOperatorStateBackend@10967} 
  128.  metrics = {OperatorMetricGroup@10968} 
  129.  latencyStats = {LatencyStats@10969} 
  130.  processingTimeService = {ProcessingTimeServiceImpl@10970} 
  131.  timeServiceManager = {InternalTimeServiceManager@10971} 
  132.  combinedWatermark = -9223372036854775808
  133.  input1Watermark = -9223372036854775808
  134.  input2Watermark = -9223372036854775808
  135.   
  136. // 处理table 1 
  138. processElement1:118, StreamingJoinOperator (org.apache.Flink.table.runtime.operators.join.stream)
  139. processRecord1:135, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  140. lambda$new$0:100, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  141. accept:-1, 169462196 (org.apache.Flink.streaming.runtime.io.StreamTwoInputProcessor$$Lambda$733)
  142. emitRecord:362, StreamTwoInputProcessor$StreamTaskNetworkOutput (org.apache.Flink.streaming.runtime.io)
  143. processElement:151, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  144. emitNext:128, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  145. processInput:182, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  146. processInput:311, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  147. runDefaultAction:-1, 1284793893 (org.apache.Flink.streaming.runtime.tasks.StreamTask$$Lambda$713)
  148. runMailboxLoop:187, MailboxProcessor (org.apache.Flink.streaming.runtime.tasks.mailbox)
  149. runMailboxLoop:487, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  150. invoke:470, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  151. doRun:707, Task (org.apache.Flink.runtime.taskmanager)
  152. run:532, Task (org.apache.Flink.runtime.taskmanager)
  153. run:748, Thread (java.lang)
  154.   
  155. // 处理table 2 
  157. processElement2:123, StreamingJoinOperator (org.apache.Flink.table.runtime.operators.join.stream)
  158. processRecord2:145, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  159. lambda$new$1:107, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  160. accept:-1, 76811487 (org.apache.Flink.streaming.runtime.io.StreamTwoInputProcessor$$Lambda$734)
  161. emitRecord:362, StreamTwoInputProcessor$StreamTaskNetworkOutput (org.apache.Flink.streaming.runtime.io)
  162. processElement:151, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  163. emitNext:128, StreamTaskNetworkInput (org.apache.Flink.streaming.runtime.io)
  164. processInput:185, StreamTwoInputProcessor (org.apache.Flink.streaming.runtime.io)
  165. processInput:311, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  166. runDefaultAction:-1, 1284793893 (org.apache.Flink.streaming.runtime.tasks.StreamTask$$Lambda$713)
  167. runMailboxLoop:187, MailboxProcessor (org.apache.Flink.streaming.runtime.tasks.mailbox)
  168. runMailboxLoop:487, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  169. invoke:470, StreamTask (org.apache.Flink.streaming.runtime.tasks)
  170. doRun:707, Task (org.apache.Flink.runtime.taskmanager)
  171. run:532, Task (org.apache.Flink.runtime.taskmanager)
  172. run:748, Thread (java.lang)
  173.   
  174. // 处理table 1   
  175.   
  176. processRecord1:134, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  177. lambda$new$0:100, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  178. accept:-1, 230607815 (org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor$$Lambda$735)
  179. emitRecord:362, StreamTwoInputProcessor$StreamTaskNetworkOutput (org.apache.flink.streaming.runtime.io)
  180. processElement:151, StreamTaskNetworkInput (org.apache.flink.streaming.runtime.io)
  181. emitNext:128, StreamTaskNetworkInput (org.apache.flink.streaming.runtime.io)
  182. processInput:182, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  183. processInput:311, StreamTask (org.apache.flink.streaming.runtime.tasks)
  184. runDefaultAction:-1, 33038573 (org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$718)
  185. runMailboxLoop:187, MailboxProcessor (org.apache.flink.streaming.runtime.tasks.mailbox)
  186. runMailboxLoop:487, StreamTask (org.apache.flink.streaming.runtime.tasks)
  187. invoke:470, StreamTask (org.apache.flink.streaming.runtime.tasks)
  188. doRun:707, Task (org.apache.flink.runtime.taskmanager)
  189. run:532, Task (org.apache.flink.runtime.taskmanager)
  190. run:748, Thread (java.lang)	
  191.   
  192. // 处理table 2   
  193.   
  194. processRecord2:144, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  195. lambda$new$1:107, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  196. accept:-1, 212261435 (org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor$$Lambda$736)
  197. emitRecord:362, StreamTwoInputProcessor$StreamTaskNetworkOutput (org.apache.flink.streaming.runtime.io)
  198. processElement:151, StreamTaskNetworkInput (org.apache.flink.streaming.runtime.io)
  199. emitNext:128, StreamTaskNetworkInput (org.apache.flink.streaming.runtime.io)
  200. processInput:185, StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)
  201. processInput:311, StreamTask (org.apache.flink.streaming.runtime.tasks)
  202. runDefaultAction:-1, 33038573 (org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$718)
  203. runMailboxLoop:187, MailboxProcessor (org.apache.flink.streaming.runtime.tasks.mailbox)
  204. runMailboxLoop:487, StreamTask (org.apache.flink.streaming.runtime.tasks)
  205. invoke:470, StreamTask (org.apache.flink.streaming.runtime.tasks)
  206. doRun:707, Task (org.apache.flink.runtime.taskmanager)
  207. run:532, Task (org.apache.flink.runtime.taskmanager)
  208. run:748, Thread (java.lang)  
  209.   
  1. import java.sql.Timestamp
  2. import org.apache.flink.api.java.utils.ParameterTool
  3. import org.apache.flink.api.scala._
  4. import org.apache.flink.streaming.api.TimeCharacteristic
  5. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  6. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  7. import org.apache.flink.streaming.api.windowing.time.Time
  8. import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
  9. import org.apache.flink.table.api.scala._
  10. import org.apache.flink.types.Row
  12. import scala.collection.mutable
  14. import java.sql.Timestamp
  16. import org.apache.flink.api.scala._
  17. import org.apache.flink.streaming.api.TimeCharacteristic
  18. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  19. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  20. import org.apache.flink.streaming.api.windowing.time.Time
  21. import org.apache.flink.table.api.TableEnvironment
  22. import org.apache.flink.table.api.scala._
  23. import org.apache.flink.types.Row
  25. import scala.collection.mutable
  27. object SimpleTimeIntervalJoinA {
  28.   def main(args: Array[String]): Unit = {
  29.     val params = ParameterTool.fromArgs(args)
  30.     val planner = if (params.has("planner")) params.get("planner") else "flink"
  32.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  34.     val tEnv = if (planner == "blink") {  // use blink planner in streaming mode
  35.       val settings = EnvironmentSettings.newInstance()
  36.         .useBlinkPlanner()
  37.         .inStreamingMode()
  38.         .build()
  39.       StreamTableEnvironment.create(env, settings)
  40.     } else if (planner == "flink") {  // use flink planner in streaming mode
  41.       StreamTableEnvironment.create(env)
  42.     } else {
  43.       System.err.println("The planner is incorrect. Please run 'StreamSQLExample --planner <planner>', " +
  44.         "where planner (it is either flink or blink, and the default is flink) indicates whether the " +
  45.         "example uses flink planner or blink planner.")
  46.       return
  47.     }
  48.     env.setParallelism(1)
  49.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  50.     // 构造订单数据
  51.     val ordersData = new mutable.MutableList[(String, String, Timestamp)]
  52.     ordersData.+=(("001", "iphone", new Timestamp(1545800002000L)))
  53.     ordersData.+=(("002", "mac", new Timestamp(1545800003000L)))
  54.     ordersData.+=(("003", "book", new Timestamp(1545800004000L)))
  55.     ordersData.+=(("004", "cup", new Timestamp(1545800018000L)))
  57.     // 构造付款表
  58.     val paymentData = new mutable.MutableList[(String, String, Timestamp)]
  59.     paymentData.+=(("001", "alipay", new Timestamp(1545803501000L)))
  60.     paymentData.+=(("002", "card", new Timestamp(1545803602000L)))
  61.     paymentData.+=(("003", "card", new Timestamp(1545803610000L)))
  62.     paymentData.+=(("004", "alipay", new Timestamp(1545803611000L)))
  63.     val orders = env
  64.       .fromCollection(ordersData)
  65.       .assignTimestampsAndWatermarks(new TimestampExtractor[String, String]())
  66.       .toTable(tEnv, 'orderId, 'productName, 'orderTime.rowtime)
  67.     val ratesHistory = env
  68.       .fromCollection(paymentData)
  69.       .assignTimestampsAndWatermarks(new TimestampExtractor[String, String]())
  70.       .toTable(tEnv, 'orderId, 'payType, 'payTime.rowtime)
  72.     tEnv.registerTable("Orders", orders)
  73.     tEnv.registerTable("Payment", ratesHistory)
  75.     var sqlQuery =
  76.       """
  77.         |SELECT
  78.         |  o.orderId,
  79.         |  o.productName,
  80.         |  p.payType,
  81.         |  o.orderTime,
  82.         |  cast(payTime as timestamp) as payTime
  83.         |FROM
  84.         |  Orders AS o left outer JOIN Payment AS p ON o.orderId = p.orderId AND
  85.         | p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
  86.         |""".stripMargin
  87.     tEnv.registerTable("TemporalJoinResult", tEnv.sqlQuery(sqlQuery))
  89.     val result = tEnv.scan("TemporalJoinResult").toAppendStream[Row]
  90.     result.print()
  91.     print(tEnv.explain(tEnv.sqlQuery(sqlQuery)))
  92.     env.execute()
  93.   }
  94. }
  96. class TimestampExtractor[T1, T2]
  97.   extends BoundedOutOfOrdernessTimestampExtractor[(T1, T2, Timestamp)](Time.seconds(10)) {
  98.   override def extractTimestamp(element: (T1, T2, Timestamp)): Long = {
  99.     element._3.getTime
  100.   }
  101. }

输出如下

  1. == Abstract Syntax Tree ==
  2. LogicalProject(orderId=[$0], productName=[$1], payType=[$4], orderTime=[$2], payTime=[CAST($5):TIMESTAMP(6)])
  3. +- LogicalJoin(condition=[AND(=($0, $3), >=($5, $2), <=($5, +($2, 3600000:INTERVAL HOUR)))], joinType=[left])
  4.    :- LogicalTableScan(table=[[default_catalog, default_database, Orders]])
  5.    +- LogicalTableScan(table=[[default_catalog, default_database, Payment]])
  7. == Optimized Logical Plan ==
  8. Calc(select=[orderId, productName, payType, orderTime, CAST(CAST(payTime)) AS payTime])
  9. +- WindowJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-3600000, leftUpperBound=0, leftTimeIndex=2, rightTimeIndex=2], where=[AND(=(orderId, orderId0), >=(payTime, orderTime), <=(payTime, +(orderTime, 3600000:INTERVAL HOUR)))], select=[orderId, productName, orderTime, orderId0, payType, payTime])
  10.    :- Exchange(distribution=[hash[orderId]])
  11.    :  +- DataStreamScan(table=[[default_catalog, default_database, Orders]], fields=[orderId, productName, orderTime])
  12.    +- Exchange(distribution=[hash[orderId]])
  13.       +- DataStreamScan(table=[[default_catalog, default_database, Payment]], fields=[orderId, payType, payTime])
  15. == Physical Execution Plan ==
  16. Stage 1 : Data Source
  17. 	content : Source: Collection Source
  19. 	Stage 2 : Operator
  20. 		content : Timestamps/Watermarks
  21. 		ship_strategy : FORWARD
  23. Stage 3 : Data Source
  24. 	content : Source: Collection Source
  26. 	Stage 4 : Operator
  27. 		content : Timestamps/Watermarks
  28. 		ship_strategy : FORWARD
  30. 		Stage 13 : Operator
  31. 			content : SourceConversion(table=[default_catalog.default_database.Orders], fields=[orderId, productName, orderTime])
  32. 			ship_strategy : FORWARD
  34. 			Stage 15 : Operator
  35. 				content : SourceConversion(table=[default_catalog.default_database.Payment], fields=[orderId, payType, payTime])
  36. 				ship_strategy : FORWARD
  38. 				Stage 17 : Operator
  39. 					content : WindowJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-3600000, leftUpperBound=0, leftTimeIndex=2, rightTimeIndex=2], where=[((orderId = orderId0) AND (payTime >= orderTime) AND (payTime <= (orderTime + 3600000:INTERVAL HOUR)))], select=[orderId, productName, orderTime, orderId0, payType, payTime])
  40. 					ship_strategy : HASH
  42. 					Stage 18 : Operator
  43. 						content : Calc(select=[orderId, productName, payType, orderTime, CAST(CAST(payTime)) AS payTime])
  44. 						ship_strategy : FORWARD
  46. 001,iphone,alipay,2018-12-26T04:53:22,2018-12-26T05:51:41
  47. 002,mac,card,2018-12-26T04:53:23,2018-12-26T05:53:22
  48. 004,cup,alipay,2018-12-26T04:53:38,2018-12-26T05:53:31
  49. 003,book,null,2018-12-26T04:53:24,null

相关类以及调用栈

  1. class StreamExecWindowJoin {  
  2. }
  4. class StreamExecWindowJoinRule
  5.   extends ConverterRule(
  6.     classOf[FlinkLogicalJoin],
  7.     FlinkConventions.LOGICAL,
  8.     FlinkConventions.STREAM_PHYSICAL,
  9.     "StreamExecWindowJoinRule") {
  10. }
  13. matches:54, StreamExecWindowJoinRule (org.apache.flink.table.planner.plan.rules.physical.stream)
  14. matchRecurse:263, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  15. match:247, VolcanoRuleCall (org.apache.calcite.plan.volcano)
  16. fireRules:1534, VolcanoPlanner (org.apache.calcite.plan.volcano)
  17. registerImpl:1807, VolcanoPlanner (org.apache.calcite.plan.volcano)
  18. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  19. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  20. ensureRegistered:90, VolcanoPlanner (org.apache.calcite.plan.volcano)
  21. onRegister:329, AbstractRelNode (org.apache.calcite.rel)
  22. registerImpl:1668, VolcanoPlanner (org.apache.calcite.plan.volcano)
  23. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  24. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  25. ensureRegistered:90, VolcanoPlanner (org.apache.calcite.plan.volcano)
  26. onRegister:329, AbstractRelNode (org.apache.calcite.rel)
  27. registerImpl:1668, VolcanoPlanner (org.apache.calcite.plan.volcano)
  28. register:846, VolcanoPlanner (org.apache.calcite.plan.volcano)
  29. ensureRegistered:868, VolcanoPlanner (org.apache.calcite.plan.volcano)
  30. changeTraits:529, VolcanoPlanner (org.apache.calcite.plan.volcano)
  31. run:324, Programs$RuleSetProgram (org.apache.calcite.tools)
  32. optimize:64, FlinkVolcanoProgram (org.apache.flink.table.planner.plan.optimize.program)
  33. apply:62, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  34. apply:58, FlinkChainedProgram$$anonfun$optimize$1 (org.apache.flink.table.planner.plan.optimize.program)
  35. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  36. apply:157, TraversableOnce$$anonfun$foldLeft$1 (scala.collection)
  37. foreach:891, Iterator$class (scala.collection)
  38. foreach:1334, AbstractIterator (scala.collection)
  39. foreach:72, IterableLike$class (scala.collection)
  40. foreach:54, AbstractIterable (scala.collection)
  41. foldLeft:157, TraversableOnce$class (scala.collection)
  42. foldLeft:104, AbstractTraversable (scala.collection)
  43. optimize:57, FlinkChainedProgram (org.apache.flink.table.planner.plan.optimize.program)
  44. optimizeTree:170, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  45. doOptimize:90, StreamCommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  46. optimize:77, CommonSubGraphBasedOptimizer (org.apache.flink.table.planner.plan.optimize)
  47. optimize:248, PlannerBase (org.apache.flink.table.planner.delegation)
  48. translate:151, PlannerBase (org.apache.flink.table.planner.delegation)
  49. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  50. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  51. toAppendStream:101, TableConversions (org.apache.flink.table.api.scala)
  52. main:93, SimpleTimeIntervalJoinA$ (spendreport)
  53. main:-1, SimpleTimeIntervalJoinA (spendreport)
  54.   
  55.   
  56. translateToPlanInternal:136, StreamExecWindowJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  57. translateToPlanInternal:53, StreamExecWindowJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  58. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  59. translateToPlan:53, StreamExecWindowJoin (org.apache.flink.table.planner.plan.nodes.physical.stream)
  60. translateToPlanInternal:54, StreamExecCalc (org.apache.flink.table.planner.plan.nodes.physical.stream)
  61. translateToPlanInternal:39, StreamExecCalc (org.apache.flink.table.planner.plan.nodes.physical.stream)
  62. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  63. translateToPlan:38, StreamExecCalcBase (org.apache.flink.table.planner.plan.nodes.physical.stream)
  64. translateToTransformation:184, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  65. translateToPlanInternal:153, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  66. translateToPlanInternal:48, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  67. translateToPlan:58, ExecNode$class (org.apache.flink.table.planner.plan.nodes.exec)
  68. translateToPlan:48, StreamExecSink (org.apache.flink.table.planner.plan.nodes.physical.stream)
  69. apply:60, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.flink.table.planner.delegation)
  70. apply:59, StreamPlanner$$anonfun$translateToPlan$1 (org.apache.flink.table.planner.delegation)
  71. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  72. apply:234, TraversableLike$$anonfun$map$1 (scala.collection)
  73. foreach:891, Iterator$class (scala.collection)
  74. foreach:1334, AbstractIterator (scala.collection)
  75. foreach:72, IterableLike$class (scala.collection)
  76. foreach:54, AbstractIterable (scala.collection)
  77. map:234, TraversableLike$class (scala.collection)
  78. map:104, AbstractTraversable (scala.collection)
  79. translateToPlan:59, StreamPlanner (org.apache.flink.table.planner.delegation)
  80. translate:153, PlannerBase (org.apache.flink.table.planner.delegation)
  81. toDataStream:210, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  82. toAppendStream:107, StreamTableEnvironmentImpl (org.apache.flink.table.api.scala.internal)
  83. toAppendStream:101, TableConversions (org.apache.flink.table.api.scala)
  84. main:93, SimpleTimeIntervalJoinA$ (spendreport)
  85. main:-1, SimpleTimeIntervalJoinA (spendreport)

Flink table&Sql中使用Calcite

Flink sql的实现

Calcite 功能简析及在 Flink 的应用

基于Flink1.8 深入理解Flink Sql执行流程 + Flink Sql语法扩展

使用Flink Table &Sql api来构建批量和流式应用(3)Flink Sql 使用

Flink关系型API： Table API 与SQL

Flink sql的实现

Flink如何实现动态表与静态表的Join操作

一文解析Flink SQL工作流程

Flink1.9-table/SQLAPI

【Flink SQL引擎】：Calcite 功能简析及在 Flink 的应用

Apache Calcite 处理流程详解（一）

Apache Calcite 优化器详解（二）

揭秘 Flink 1.9 新架构，Blink Planner 你会用了吗？

Flink 原理与实现：Table & SQL API | Jark’s Blog

版权声明：本文为rossiXYZ原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/rossiXYZ/p/12770436.html