本文将从作业配置和Flink SQL优化两方面为您介绍如何提升Flink SQL作业性能。
作业配置优化推荐方案
资源优化技巧
VVP中限制了JobManager和TaskManager的CPU的实际使用大小，配置了多少个CPU，最大就只能使用多少个CPU。因此在资源优化时，建议：
作业并发大时：在作业的部署详情页签的资源配置中，增加JobManager的资源，提高CPU和内存的大小，例如：
Job Manager CPUs设置为4。
Job Manager Memory设置为8 GiB。
作业拓扑较复杂时，在作业的部署详情页签的资源配置中，增加TaskManager的资源，提高CPU和内存的大小，例如：
Task Manager CPUs设置为2。
Task Manager Memory设置为4 GiB。
不建议修改taskmanager.numberOfTaskSlots，保持默认值1。
提升吞吐和解决数据热点的推荐配置
在其他配置中添加以下代码，具体操作请参见如何配置自定义的作业运行参数？和Group Aggregate优化技巧。
execution.checkpointing.interval: 180s table.exec.state.ttl: 129600000 table.exec.mini-batch.enabled: true table.exec.mini-batch.allow-latency: 5s table.optimizer.distinct-agg.split.enabled: true参数解释如下表所示。
参数
说明
execution.checkpointing.interval
Checkpoint间隔时间，单位为毫秒。
state.backend
State backend的配置。
table.exec.state.ttl
State数据的生命周期，单位为毫秒。
table.exec.mini-batch.enabled
是否开启minibatch。
table.exec.mini-batch.allow-latency
批量输出数据的时间间隔。
table.exec.mini-batch.size
微批操作所缓存的最大数据条数。
说明实时计算引擎VVR已对Minibatch机制进行了优化，建议不设置该参数，具体参考重要参数说明。
table.optimizer.distinct-agg.split.enabled
是否开启PartialFinal优化，解决COUNT DISTINCT热点问题。
提升双流Join类型作业的性能配置
流式SQL中双流Join算子支持自动推导开启KV分离优化。在实时计算引擎VVR 6.0.1及以上版本中，SQL作业双流Join算子会根据作业特点，自动推导并开启State KV分离优化功能，无需您额外配置。开启State KV分离优化功能后，可以显著提升双流Join类型作业的性能。在典型场景的性能测试中，有40%以上的性能提升。
您可以通过table.exec.join.kv-separate配置项对该功能进行显式控制，参数取值详情如下：
AUTO（默认值）：表示引擎内部会根据双流Join算子的State特点自动开启。
FORCE：表示强制开启KV分离优化。
NONE：表示强制关闭KV分离优化。
说明该功能仅对Gemini StateBackend生效。
Flink SQL优化推荐方案
Group Aggregate优化技巧
开启MiniBatch（提升吞吐）
MiniBatch是缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐并减少数据的输出量。
MiniBatch主要基于事件消息来触发微批处理，事件消息会按您指定的时间间隔在源头插入。
适用场景
微批处理通过增加延迟换取高吞吐，如果您有超低延迟的要求，不建议开启微批处理。对于一般聚合场景，微批处理可以显著提升系统性能，建议开启。
开启方式
MiniBatch默认关闭，在其他配置中填写以下代码即可开启，具体操作请参见如何配置自定义的作业运行参数？。
table.exec.mini-batch.enabled: true table.exec.mini-batch.allow-latency: 5s参数解释如下表所示。
参数
说明
table.exec.mini-batch.enabled
是否开启mini-batch。
table.exec.mini-batch.allow-latency
批量输出数据的时间间隔。
table.exec.mini-batch.size
微批操作所缓存的最大数据条数。
说明该参数需要同时与上述两个参数一起使用才会生效。实时计算引擎VVR已对Minibatch机制进行了优化，建议不设置该参数，具体参考重要参数说明。
开启LocalGlobal（解决常见数据热点问题）
LocalGlobal本质上能够靠LocalAgg的聚合筛除部分倾斜数据，从而降低GlobalAgg的热点，提升性能。
LocalGlobal优化将原先的Aggregate分成Local和Global两阶段聚合，即MapReduce模型中的Combine和Reduce两阶段处理模式。第一阶段在上游节点积攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。
适用场景
提升普通聚合（例如SUM、COUNT、MAX、MIN和AVG）的性能，以及这些场景下的数据热点问题。
使用限制
LocalGlobal是默认开启的，但是有以下限制：
在minibatch开启的前提下才能生效。
需要使用AggregateFunction实现Merge。
判断是否生效
观察最终生成的拓扑图的节点名字中是否包含GlobalGroupAggregate或LocalGroupAggregate。
开启PartialFinal（解决COUNT DISTINCT热点问题）
为了解决COUNT DISTINCT的热点问题，通常需要手动改写为两层聚合（增加按Distinct Key取模的打散层）。目前，实时计算提供了COUNT DISTINCT自动打散，即PartialFinal优化，您无需自行改写为两层聚合。
LocalGlobal优化针对普通聚合（例如SUM、COUNT、MAX、MIN和AVG）有较好的效果，对于COUNT DISTINCT收效不明显，因为COUNT DISTINCT在Local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点问题。
适用场景
使用COUNT DISTINCT，但无法满足聚合节点性能要求。
说明不能在包含UDAF的Flink SQL中使用PartialFinal优化方法。
数据量较少的情况，不建议使用PartialFinal优化方法，浪费资源。因为PartialFinal优化会自动打散成两层聚合，引入额外的网络Shuffle。
开启方式
默认不开启。如果您需要开启，在其他配置中填写以下代码，具体操作请参见如何配置自定义的作业运行参数？。
table.optimizer.distinct-agg.split.enabled: true判断是否生效
观察最终生成的拓扑图，是否由原来一层的聚合变成了两层的聚合。
AGG WITH CASE WHEN改写为AGG WITH FILTER语法（提升大量COUNT DISTINCT场景性能）
统计作业需要计算各种维度的UV，例如全网UV、来自手机客户端的UV、来自PC的UV等等。建议使用标准的AGG WITH FILTER语法来代替CASE WHEN实现多维度统计的功能。实时计算目前的SQL优化器能分析出Filter参数，从而同一个字段上计算不同条件下的COUNT DISTINCT能共享State，减少对State的读写操作。性能测试中，使用AGG WITH FILTER语法来代替CASE WHEN能够使性能提升1倍。
适用场景
对于同一个字段上计算不同条件下的COUNT DISTINCT结果的场景，性能提升很大。
原始写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2优化写法
COUNT(distinct visitor_id) as UV1 , COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2
使用Delta Aggregation（无状态聚合）
说明仅实时计算引擎 VVR 11.8 及以上版本支持 Delta Aggregation。
Delta Aggregation 是 Flink 面向流式非窗口分组聚合提供的无状态聚合能力。与普通分组聚合在 Flink State 中为每个分组持续维护聚合累加器不同，Delta Aggregation 在收到数据变更后，使用分组键通过异步 Lookup 获取该分组的当前明细，并基于这些明细重新计算聚合结果。
这种聚合方式使 Flink 侧的聚合状态规模不再由历史分组数量决定。作业无需为不断增加的分组持续扩展聚合状态，可以减少 Checkpoint 需要持久化的数据量、缩短作业恢复时间，并使大规模分组聚合作业的资源使用更加稳定。
适用场景
Delta Aggregation 通过异步 Lookup 获取分组的当前明细，并按需重新计算聚合结果，适用于希望实现以下目标之一的作业：
当作业逻辑修改后，普通分组聚合需无状态启动流作业并重建状态；Delta Aggregation 可先用无状态批作业更新结果表，再有状态启动流作业，利用批处理的高吞吐能力加快结果回填和作业恢复。
分组状态规模较大时，降低 Checkpoint 延迟并缓解由状态处理引起的反压。
作业重启时，大幅降低需要恢复的聚合状态规模，缩短恢复时间。
减少聚合状态占用的计算和存储资源，降低计算资源（CU）成本。
多个作业基于同一份明细数据进行聚合时，共享源表中的当前明细状态。
需要进行中间结果探查时，可以直接查询源表中的当前明细，便于验证聚合结果和定位问题。
使用限制
查询结构
仅支持流作业中的单层非窗口分组聚合，不支持窗口聚合和级联聚合。
GROUP BY中必须包含一组可直接映射到源表字段的列，并完整包含源表主键或一个可查询索引。不支持使用
PROCTIME()、RAND()等非确定性表达式。
上下游表
源表 Connector 必须同时支持流式 Scan 和异步 Lookup。
源表不能定义 Watermark。
Delta Aggregation 向下游发送的变更数据仅包含
UPDATE_AFTER（+U）和DELETE（-D），不包含UPDATE_BEFORE（-U）。
聚合函数
支持
SUM、COUNT、AVG、MIN、MAX、LISTAGG、FIRST_VALUE和LAST_VALUE。暂不支持
DISTINCT、带FILTER子句的聚合、UDAF、Python 聚合函数及其他未列出的聚合函数。
示例
某电商平台需要为海量订单维护当前记录数和金额，结果写入汇总表，供订单查询服务按商家和订单直接读取。以下示例使用完整源表主键作为 Lookup Key。
步骤一：创建源表和结果表
源表必须支持流式 Scan 和异步 Lookup。
CREATE TABLE orders ( merchant_id BIGINT, -- 商家 ID order_id BIGINT, -- 订单 ID amount DECIMAL(18, 2), -- 订单当前金额 PRIMARY KEY (merchant_id, order_id) NOT ENFORCED ) WITH ( 'connector' = '<source-connector>', '<source-option>' = '<source-option-value>' );结果表使用全部分组字段作为主键。
CREATE TABLE order_current_summary ( merchant_id BIGINT, -- 商家 ID order_id BIGINT, -- 订单 ID current_record_count BIGINT, -- 当前有效记录数 total_amount DECIMAL(38, 2), -- 当前汇总金额 PRIMARY KEY (merchant_id, order_id) NOT ENFORCED ) WITH ( 'connector' = '<sink-connector>', '<sink-option>' = '<sink-option-value>' );步骤二：运行聚合作业
在
INSERT INTO语句之前开启 Delta Aggregation：SET 'table.optimizer.delta-agg.strategy' = 'EVENTUAL';使用普通分组聚合 SQL，无需引入专用语法：
INSERT INTO order_current_summary SELECT merchant_id, order_id, COUNT(*) AS current_record_count, SUM(amount) AS total_amount FROM orders GROUP BY merchant_id, order_id;本例的
GROUP BY (merchant_id, order_id)完整覆盖源表主键，结果表主键也与分组字段一致。步骤三：验证执行计划
提交作业前查看 SQL 执行计划。计划中出现
DeltaAggregate节点，表示 Delta Aggregation 已生效，例如：DeltaAggregate( groupBy=[merchant_id, order_id], lookupKeys=[merchant_id, order_id] )
可选调优
建议先使用默认参数运行作业，再根据 Lookup 性能和作业吞吐逐项调整。以下
SET语句均应放在对应的INSERT INTO语句之前。选择启用策略
table.optimizer.delta-agg.strategy支持以下取值：取值
说明
EVENTUAL强制使用 Delta Aggregation；查询不满足条件时，作业会在生成执行计划阶段报错。
AUTO尝试使用 Delta Aggregation；查询不满足条件时，回退到普通分组聚合。
NONE关闭 Delta Aggregation，默认值。
调整异步 Lookup 参数
可以通过以下参数调整异步 Lookup 的并发量和超时时间：
SET 'table.exec.async-lookup.buffer-capacity' = '100'; SET 'table.exec.async-lookup.timeout' = '3 min';参数
默认值
说明
table.exec.async-lookup.buffer-capacity100每个并发子任务允许同时执行的异步 Lookup 请求数。增大该值可提高异步 Lookup 的并发能力。
table.exec.async-lookup.timeout3 min单次异步处理的超时时间。
如需进一步了解异步 Lookup 的调优方法，请参见维表 JOIN 语句。
开启 MiniBatch Delta Aggregation
Delta Aggregation 与 Group Aggregation 均支持 MiniBatch 优化。对于 Delta Aggregation，MiniBatch 会合并同一批次内相同分组键的变更，使每个分组键在一个批次内最多发起一次 Lookup，从而减少源表 Lookup 次数。
可通过以下配置为 Delta Aggregation 开启 MiniBatch：
SET 'table.exec.mini-batch.enabled' = 'true'; SET 'table.exec.mini-batch.allow-latency' = '10s'; SET 'table.exec.mini-batch.size' = '1000';以上配置表示：每个并发子任务累计收到 1000 条输入，或者批次等待达到 10 秒时触发处理。
如需进一步了解 MiniBatch 的优化原理和参数说明，请参见本节前文的“开启MiniBatch（提升吞吐）”小节。
开启 Delta Aggregation Cache
如果同一分组键会被反复更新，可以开启 Cache 复用已经查询到的分组明细。缓存命中时，Flink 根据输入变更更新缓存，并直接重新计算聚合结果，无需再次发起 Lookup。
开启 Cache 后，源表必须定义主键。Flink 使用源表主键标识和更新缓存中的明细记录；
table.exec.delta-agg.cache-size必须大于0。SET 'table.exec.delta-agg.cache-enabled' = 'true'; SET 'table.exec.delta-agg.cache-size' = '2000';参数
默认值
说明
table.exec.delta-agg.cache-enabledfalse是否开启 Delta Aggregation Cache。
table.exec.delta-agg.cache-size10000每个 Delta Aggregation 并发子任务最多缓存的分组键数量，仅在开启 Cache 时生效。
放宽聚合结果过滤条件校验
Delta Aggregation 不会发送旧聚合结果的
-U。以下游过滤条件HAVING SUM(amount) >= 100为例，当某个分组的SUM(amount)从120更新为80时：+U(..., 120) → 满足条件，写入结果表 +U(..., 80) → 不满足条件，被过滤由于没有
-U撤回旧结果，结果表中仍会保留SUM(amount)=120的记录。为避免该问题，Delta Aggregation 默认要求下游过滤条件只能引用聚合结果的唯一键字段（通常为GROUP BY字段）。如果过滤条件引用
SUM等非唯一键字段，EVENTUAL策略会报错，AUTO策略会回退到普通分组聚合。GROUP BY之前的WHERE条件不受此校验影响。如果业务能够容忍上述影响，可以通过以下配置跳过该校验：
SET 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true';参数
默认值
说明
table.optimizer.delta-agg.ignore-non-unique-key-filterfalse是否跳过 Delta Aggregation 聚合结果的非唯一键过滤条件校验。
该配置仅跳过校验，不会改变 SQL 中的过滤条件，也不会使 Delta Aggregation 生成
-U。因此，Flink 不会补发用于清理上述旧结果的撤回消息。
Join优化技巧
开启MiniBatch（提升吞吐）
MiniBatch是缓存一定的数据后再触发处理，在一批缓存中会尝试合并的消息，以此减少处理的数据和State访问，从而提升吞吐并减少中间结果的输出，进而减轻下游的处理负载。MiniBatch在以下条件下会触发微批处理：1. 基于配置的最大允许时间间隔 2. 到达缓存的批大小 3. 其他比如cp 触发。
适用场景
微批处理通过一定的处理延迟来换取更高吞吐，如果您有超低延迟的要求，不建议开启微批处理。在常见的Join场景中，尤其涉及级联Outer Join（如以下SQL示例）时，启用微批处理能显著提升系统性能。其原理在于：上游算子通过消息折叠/合并机制，有效抑制了消息在下游算子中的放大效应。
SELECT a.id as a_id, a.a_content, B.id as b_id, B.b_content FROM a LEFT JOIN (SELECT * FROM b LEFT JOIN c on b.prd_id = c.id) B ON a.id = B.id
开启方式
VVR-8.0.4及以上版本支持此优化，MiniBatch默认关闭，在其他配置中填写以下代码即可开启，具体操作请参见如何配置自定义的作业运行参数？。
table.exec.mini-batch.enabled: true table.exec.mini-batch.allow-latency: 5s table.exec.stream.join.mini-batch-enabled: true
TopN优化技巧
TopN算法
当TopN的输入是非更新流（例如SLS数据源），TopN只有1种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG或JOIN计算），TopN有2种算法，性能从高到低分别是：UpdateFastRank和RetractRank。算法名字会显示在拓扑图的节点名字上。
AppendRank：对于非更新流，只支持该算法。
UpdateFastRank：对于更新流，最优算法。
RetractRank：对于更新流，保底算法。性能不佳，在某些业务场景下可优化成UpdateFastRank。
下面介绍RetractRank如何能优化成UpdateFastRank。使用UpdateFastRank算法需要具备3个条件：
输入流为更新流，但不能包含DELETE（D）、UPDATE_BEFORE（UB）类型的消息，否则会影响排序字段的单调性。关于输入流的消息类型，可以通过执行
EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set>命令来获取对应节点输出的消息类型，语法详情请参见EXPLAIN语句。输入流有Primary Key信息，例如上游做了GROUP BY聚合操作。
排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC。
如果您要获取到UpdateFastRank的优化Plan，则您需要在使用ORDER BY SUM DESC时，添加SUM为正数的过滤条件，确保total_fee为正数。
说明如下示例中的random_test表为非更新流，对应的GROUP分组聚合的结果不会包含DELETE（D）、UPDATE_BEFORE（UB）消息，所以对应的聚合结果字段才能保持单调性。
可以优化成UpdateFastRank的示例：
insert into print_test SELECT cate_id, seller_id, stat_date, pay_ord_amt -- 不输出rownum字段，能减小结果表的输出量。 FROM ( SELECT *, ROW_NUMBER () OVER ( -- 注意：PARTITION BY的列要被子查询中的GROUP BY分组聚合字段包含；另外要有时间字段，否则State过期会导致数据错乱。 PARTITION BY cate_id, stat_date ORDER BY pay_ord_amt DESC ) as rownum --根据上游sum结果排序。 FROM ( SELECT cate_id, seller_id, stat_date, -- 重点：声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据。 sum (total_fee) filter ( where total_fee >= 0 ) as pay_ord_amt FROM random_test WHERE total_fee >= 0 GROUP BY seller_id, stat_date, cate_id ) a ) WHERE rownum <= 100;TopN优化方法
无排名优化
TopN的输出结果不需要显示rownum值，仅需在最终前端显示时进行1次排序，极大地减少输入结果表的数据量。无排名优化方法详情请参见Top-N。
增加TopN的Cache大小
TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式如下。
cache_hit = cache_size*parallelism/top_n/partition_key_num例如，Top100配置缓存10000条，并发50，当您的PartitionBy的Key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），观察state seek metric可能会有很多毛刺。性能会大幅下降。
因此当partitionKey维度特别大时，可以适当加大TopN的cache size，相对应的也建议适当加大TopN节点的heap memory，详情请参见配置作业部署信息。
table.exec.rank.topn-cache-size: 200000默认10000条，调整TopN cache到200000，那么理论命中率能达到200000*50/100/100000 = 100%。
PartitionBy的字段中要有时间类字段
例如每天的排名，要带上Day字段，否则TopN的最终结果会由于State TTL产生错乱。
高效去重方案
实时计算的源数据在部分场景中存在重复数据，去重成为了用户经常反馈的需求。实时计算有保留第一条（Deduplicate Keep FirstRow）和保留最后一条（Deduplicate Keep LastRow）2种去重方案。
语法
由于SQL上没有直接支持去重的语法，还要灵活地保留第一条或保留最后一条。因此我们使用了SQL的ROW_NUMBER OVER WINDOW功能来实现去重语法。去重本质上是一种特殊的TopN。
SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY col1[, col2..] ORDER BY timeAttributeCol [asc|desc]) AS rownum FROM table_name) WHERE rownum = 1参数
说明
ROW_NUMBER()
计算行号的OVER窗口函数。行号从1开始计算。
PARTITION BY col1[, col2..]
可选。指定分区的列，即去重的KEYS。
ORDER BY timeAttributeCol [asc|desc])
指定排序的列，必须是一个时间属性的字段（即Proctime或Rowtime）。可以指定顺序（Keep FirstRow）或者倒序 （Keep LastRow）。
rownum
仅支持
rownum=1或rownum<=1。如上语法所示，去重需要两层Query：
使用
ROW_NUMBER()窗口函数来对数据根据时间属性列进行排序并标上排名。当排序字段是Proctime列时，Flink就会按照系统时间去重，其每次运行的结果是不确定的。
当排序字段是Rowtime列时，Flink就会按照业务时间去重，其每次运行的结果是确定的。
对排名进行过滤，只取第一条，达到了去重的目的。
排序方向可以是按照时间列的顺序，也可以是倒序：
Deduplicate Keep FirstRow：顺序并取第一条行数据。
Deduplicate Keep LastRow：倒序并取第一条行数据。
Deduplicate Keep FirstRow
保留首行的去重策略：保留KEY下第一条出现的数据，之后出现该KEY下的数据会被丢弃掉。因为STATE中只存储了KEY数据，所以性能较优，示例如下。
SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) as rowNum FROM T ) WHERE rowNum = 1以上示例是将表T按照字段b进行去重，并按照系统时间保留第一条数据。proctime在这里是源表T中的一个具有Processing Time属性的字段。如果您按照系统时间去重，也可以将proctime字段简化proctime()函数调用，可以省略proctime字段的声明。
Deduplicate Keep LastRow
保留末行的去重策略：保留KEY下最后一条出现的数据。保留末行的去重策略性能略优于LAST_VALUE函数，示例如下。
SELECT * FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) as rowNum FROM T ) WHERE rowNum = 1以上示例是将T表按照b和d字段进行去重，并按照业务时间保留最后一条数据。rowtime在这里是源表T中的一个具有Event Time属性的字段。
高效的内置函数
在使用内置函数时，您需要注意以下几点：
使用内置函数替换自定义函数
实时计算的内置函数在持续的优化当中，请尽量使用内置函数替换自定义函数。实时计算对内置函数主要进行了如下优化：
优化数据序列化和反序列化的耗时。
新增直接对字节单位进行操作的功能。
KEY VALUE函数使用单字符的分隔符
KEY VALUE的签名：
KEYVALUE(content, keyValueSplit, keySplit, keyName)，当keyValueSplit和KeySplit是单字符，例如，冒号（:）、逗号（,）时，系统会使用优化算法，在二进制数据上直接寻找所需的keyName值，而不会将整个content进行切分，性能约提升30%。LIKE操作注意事项
如果需要进行StartWith操作，使用
LIKE 'xxx%'。如果需要进行EndWith操作，使用
LIKE '%xxx'。如果需要进行Contains操作，使用
LIKE '%xxx%'。如果需要进行Equals操作，使用
LIKE 'xxx'，等价于str = 'xxx'。如果需要匹配下划线（_），请注意要完成转义
LIKE '%seller/_id%' ESCAPE '/'。下划线（_）在SQL中属于单字符通配符，能匹配任何字符。如果声明为LIKE '%seller_id%'，则不单会匹配seller_id，还会匹配seller#id、sellerxid或seller1id等，导致结果错误。
慎用正则函数（REGEXP）
正则表达式是非常耗时的操作，对比加减乘除通常有百倍的性能开销，而且正则表达式在某些极端情况下可能会进入无限循环，导致作业阻塞，因此建议使用LIKE。正则函数包括：
SQL Hints
为了更加灵活地提升引擎的优化能力Flink支持了SQL提示（SQL Hints），SQL提示一般可以用于以下场景：
修改执行计划：使用SQL提示，您可以更好地控制执行计划。
增加元数据（或者统计信息）：一些统计数据对于查询来说是动态的，例如已扫描的表索引、一些shuffle keys的倾斜信息等，从planner获得的计划元数据可能不准确，此时可以使用提示来配置它们。
动态表配置选项：动态表选项允许用户动态地指定或覆盖表选项，这些选项可以在每个查询的每个表范围内灵活地指定。
查询提示（Query Hints）是SQL提示的一种，用于为优化器修改执行计划提供建议，该修改只能在当前查询提示所在的查询块中生效（Query block）。 目前查询提示只支持联接提示（Join Hints）。
语法
Flink中的查询提示语法与Apache Calcite的语法一致。
# Query hints: SELECT /*+ hint [, hint ] */ ... hint: hintName '(' hintOption [, hintOption ]* ')' hintOption: simpleIdentifier | numericLiteral | stringLiteral联接提示
联接提示（Join Hints）是查询提示的一种，该提示允许您动态的优化Join，目前支持维表JOIN Hints和双流JOIN hints。