EXPLAIN 语句

EXPLAIN 语句用于解释 query 或 INSERT 语句的执行逻辑，也用于优化 query 语句的查询计划。

执行 EXPLAIN 语句

Java

可以使用 TableEnvironment 的 executeSql() 方法执行 EXPLAIN 语句。如果 EXPLAIN 操作执行成功，executeSql() 方法会返回解释结果，否则会抛出异常。

以下示例展示了如何在 TableEnvironment 中执行一条 EXPLAIN 语句。

Scala

可以使用 TableEnvironment 的 executeSql() 方法执行 EXPLAIN 语句。如果 EXPLAIN 操作执行成功，executeSql() 方法会返回解释结果，否则会抛出异常。

以下示例展示了如何在 TableEnvironment 中执行一条 EXPLAIN 语句。

Python

可以使用 TableEnvironment 的 execute_sql() 方法执行 EXPLAIN 语句。如果 EXPLAIN 操作执行成功，execute_sql() 方法会返回解释结果，否则会抛出异常。

以下示例展示了如何在 TableEnvironment 中执行一条 EXPLAIN 语句。

SQL CLI

EXPLAIN 语句可以在 SQL CLI 中执行。

以下示例展示了如何在 SQL CLI 中执行一条 EXPLAIN 语句。

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  4. // 注册名为 “Orders” 的表
  5. tEnv.executeSql("CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')");
  6. tEnv.executeSql("CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')");
  8. // 调用 TableEnvironment.explainSql() 来解释 SELECT 语句
  9. String explanation =
  10.                 tEnv.explainSql(
  11.                         "SELECT `count`, COUNT(word) FROM ("
  12.                                 + "MyTable1 WHERE word LIKE 'F%' "
  13.                                 + "UNION ALL "
  14.                                 + "SELECT `count`, word FROM MyTable2) tmp"
  15.                                 + "GROUP BY `count`");
  16. System.out.println(explanation);
  18. // 调用 TableEnvironment.executeSql() 来解释 SELECT 语句
  19. TableResult tableResult =
  20.                 tEnv.executeSql(
  21.                         "EXPLAIN PLAN FOR "
  22.                                 + "SELECT `count`, COUNT(word) FROM ("
  23.                                 + "MyTable1 WHERE word LIKE 'F%' "
  24.                                 + "UNION ALL "
  25.                                 + "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`");
  26. tableResult.print();
  28. TableResult tableResult2 =
  29.                 tEnv.executeSql(
  30.                         "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, PLAN_ADVICE, JSON_EXECUTION_PLAN "
  31.                                 + "SELECT `count`, COUNT(word) FROM ("
  32.                                 + "MyTable1 WHERE word LIKE 'F%' "
  33.                                 + "UNION ALL "
  34.                                 + "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`");
  35. tableResult2.print();

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment()
  2. val tEnv = StreamTableEnvironment.create(env)
  4. // 注册名为 “Orders” 的表
  5. tEnv.executeSql("CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  6. tEnv.executeSql("CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  8. // 调用 TableEnvironment.explainSql() 来解释 SELECT 语句
  9. val explanation = tEnv.explainSql(
  10. """
  11.   |SELECT `count`, COUNT(word)
  12.   |FROM (
  13.   |  SELECT `count`, word FROM MyTable1
  14.   |  WHERE word LIKE 'F%'
  15.   |  UNION ALL 
  16.   |  SELECT `count`, word FROM MyTable2 ) tmp
  17.   |GROUP BY `count`
  18.   |""".stripMargin)
  19. println(explanation)
  21. // 调用 TableEnvironment.executeSql() 来解释 SELECT 语句
  22. val tableResult = tEnv.executeSql(
  23. """
  24.   |EXPLAIN PLAN FOR
  25.   |SELECT `count`, COUNT(word)
  26.   |FROM (
  27.   |  SELECT `count`, word FROM MyTable1
  28.   |  WHERE word LIKE 'F%'
  29.   |  UNION ALL
  30.   |  SELECT `count`, word FROM MyTable2 ) tmp
  31.   |GROUP BY `count`
  32.   |""".stripMargin)
  33. tableResult.print()
  35. val tableResult2 = tEnv.executeSql(
  36. """
  37.   |EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, PLAN_ADVICE, JSON_EXECUTION_PLAN
  38.   |SELECT `count`, COUNT(word)
  39.   |FROM (
  40.   |  SELECT `count`, word FROM MyTable1
  41.   |  WHERE word LIKE 'F%'
  42.   |  UNION ALL
  43.   |  SELECT `count`, word FROM MyTable2 ) tmp
  44.   |GROUP BY `count`
  45.   |""".stripMargin)
  46. tableResult2.print()

Python

  1. settings = EnvironmentSettings.new_instance()...
  2. table_env = StreamTableEnvironment.create(env, settings)
  4. t_env.execute_sql("CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  5. t_env.execute_sql("CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  7. # 调用 TableEnvironment.explain_sql() 来解释 SELECT 语句
  8. explanation1 = t_env.explain_sql(
  9.     "SELECT `count`, COUNT(word) FROM (" 
  10.     "MyTable1 WHERE word LIKE 'F%' "
  11.     "UNION ALL "
  12.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  13. print(explanation1)
  15. # 调用 TableEnvironment.execute_sql() 来解释 SELECT 语句
  16. table_result = t_env.execute_sql(
  17.     "EXPLAIN PLAN FOR "
  18.     "SELECT `count`, COUNT(word) FROM (" 
  19.     "MyTable1 WHERE word LIKE 'F%' "
  20.     "UNION ALL "
  21.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  22. table_result.print()
  24. table_result2 = t_env.execute_sql(
  25.     "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, PLAN_ADVICE, JSON_EXECUTION_PLAN "
  26.     "SELECT `count`, COUNT(word) FROM (" 
  27.     "MyTable1 WHERE word LIKE 'F%' "
  28.     "UNION ALL "
  29.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  30. table_result2.print()

SQL CLI

  1. Flink SQL> CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen');
  2. [INFO] Table has been created.
  4. Flink SQL> CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen');
  5. [INFO] Table has been created.
  7. Flink SQL> EXPLAIN PLAN FOR SELECT `count`, COUNT(word) FROM
  8. > ( SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' 
  9. >   UNION ALL 
  10. >   SELECT `count`, word FROM MyTable2 ) tmp GROUP BY `count`;
  12. Flink SQL> EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, PLAN_ADVICE, JSON_EXECUTION_PLAN SELECT `count`, COUNT(word) FROM
  13. > ( SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' 
  14. >   UNION ALL 
  15. >   SELECT `count`, word FROM MyTable2 ) tmp GROUP BY `count`;

EXPLAIN 的结果如下:

EXPLAIN PLAN

  1. == Abstract Syntax Tree ==
  2. LogicalAggregate(group=[{0}], EXPR$1=[COUNT($1)])
  3. +- LogicalUnion(all=[true])
  4.    :- LogicalProject(count=[$0], word=[$1])
  5.    :  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  6.    :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
  7.    +- LogicalProject(count=[$0], word=[$1])
  8.       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
  10. == Optimized Physical Plan ==
  11. GroupAggregate(groupBy=[count], select=[count, COUNT(word) AS EXPR$1])
  12. +- Exchange(distribution=[hash[count]])
  13.    +- Union(all=[true], union=[count, word])
  14.       :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  15.       :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  16.       +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])
  18. == Optimized Execution Plan ==
  19. GroupAggregate(groupBy=[count], select=[count, COUNT(word) AS EXPR$1])
  20. +- Exchange(distribution=[hash[count]])
  21.    +- Union(all=[true], union=[count, word])
  22.       :- Calc(select=[count, word], where=[LIKE(word, 'F%')])
  23.       :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  24.       +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])

EXPLAIN PLAN WITH DETAILS

  1. == Abstract Syntax Tree ==
  2. LogicalAggregate(group=[{0}], EXPR$1=[COUNT($1)])
  3. +- LogicalUnion(all=[true])
  4.    :- LogicalProject(count=[$0], word=[$1])
  5.    :  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  6.    :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
  7.    +- LogicalProject(count=[$0], word=[$1])
  8.       +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
  10. == Optimized Physical Plan With Advice ==
  11. GroupAggregate(advice=[1], groupBy=[count], select=[count, COUNT(word) AS EXPR$1], changelogMode=[I,UA]): rowcount = 1.05E8, cumulative cost = {5.2E8 rows, 1.805E10 cpu, 4.0E9 io, 2.1E9 network, 0.0 memory}
  12. +- Exchange(distribution=[hash[count]], changelogMode=[I]): rowcount = 1.05E8, cumulative cost = {4.15E8 rows, 1.7945E10 cpu, 4.0E9 io, 2.1E9 network, 0.0 memory}
  13.    +- Union(all=[true], union=[count, word], changelogMode=[I]): rowcount = 1.05E8, cumulative cost = {3.1E8 rows, 3.05E8 cpu, 4.0E9 io, 0.0 network, 0.0 memory}
  14.       :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')], changelogMode=[I]): rowcount = 5000000.0, cumulative cost = {1.05E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}
  15.       :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word], changelogMode=[I]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}
  16.       +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word], changelogMode=[I]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}
  18. advice[1]: [ADVICE] You might want to enable local-global two-phase optimization by configuring ('table.exec.mini-batch.enabled' to 'true', 'table.exec.mini-batch.allow-latency' to a positive long value, 'table.exec.mini-batch.size' to a positive long value).
  20. == Optimized Execution Plan ==
  21. GroupAggregate(groupBy=[count], select=[count, COUNT(word) AS EXPR$1])
  22. +- Exchange(distribution=[hash[count]])
  23.    +- Union(all=[true], union=[count, word])
  24.       :- Calc(select=[count, word], where=[LIKE(word, 'F%')])
  25.       :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  26.       +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])
  28. == Physical Execution Plan ==
  29. {
  30.   "nodes" : [ {
  31.     "id" : 17,
  32.     "type" : "Source: MyTable1[15]",
  33.     "pact" : "Data Source",
  34.     "contents" : "[15]:TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
  35.     "parallelism" : 1
  36.   }, {
  37.     "id" : 18,
  38.     "type" : "Calc[16]",
  39.     "pact" : "Operator",
  40.     "contents" : "[16]:Calc(select=[count, word], where=[LIKE(word, 'F%')])",
  41.     "parallelism" : 1,
  42.     "predecessors" : [ {
  43.       "id" : 17,
  44.       "ship_strategy" : "FORWARD",
  45.       "side" : "second"
  46.     } ]
  47.   }, {
  48.     "id" : 19,
  49.     "type" : "Source: MyTable2[17]",
  50.     "pact" : "Data Source",
  51.     "contents" : "[17]:TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
  52.     "parallelism" : 1
  53.   }, {
  54.     "id" : 22,
  55.     "type" : "GroupAggregate[20]",
  56.     "pact" : "Operator",
  57.     "contents" : "[20]:GroupAggregate(groupBy=[count], select=[count, COUNT(word) AS EXPR$1])",
  58.     "parallelism" : 1,
  59.     "predecessors" : [ {
  60.       "id" : 18,
  61.       "ship_strategy" : "HASH",
  62.       "side" : "second"
  63.     }, {
  64.       "id" : 19,
  65.       "ship_strategy" : "HASH",
  66.       "side" : "second"
  67.     } ]
  68.   } ]
  69. }

ExplainDetails

使用指定的 ExplainDetail 类型来打印语句的计划。

ESTIMATED_COST

指定 ESTIMATED_COST 将使得优化器（optimizer）将估算出的成本信息附加在每个物理节点上输出。

  1. == Optimized Physical Plan ==
  2. TableSourceScan(..., cumulative cost ={1.0E8 rows, 1.0E8 cpu, 2.4E9 io, 0.0 network, 0.0 memory})

CHANGELOG_MODE

指定 CHANGELOG_MODE 将使得优化器（optimizer）将 changelog mode 附加在每个物理节点上输出。 关于 changelog mode 更多信息请参阅 动态表。

  1. == Optimized Physical Plan ==
  2. GroupAggregate(..., changelogMode=[I,UA,D])

PLAN_ADVICE

从 Flink 1.17 版本开始支持 PLAN_ADVICE。

指定 PLAN_ADVICE 将使得优化器（optimizer）分析优化后的物理执行计划并提供潜在的数据风险预警或性能调优建议。 此时输出标题将会从 “Optimized Physical Plan” 变为 “Optimized Physical Plan with Advice” 作为提示。

针对物理计划的建议按照 类型 和 范围 来区分。

建议类型说明
WARNING给出潜在的数据正确性风险
ADVICE给出可能的 SQL 调优建议
建议范围说明
QUERY_LEVEL针对整个 SQL 的建议
NODE_LEVEL针对单个物理节点的建议

PLAN_ADVICE 提供针对如下问题的建议

若检测到分组聚合可以启用两阶段优化但未开启时，优化器（optimizer）将会把建议 id 附在 GroupAggregate 节点内作为索引，在最后附上建议内容。

SQL1

  1. SET 'table.exec.mini-batch.enabled' = 'true';
  2. SET 'table.exec.mini-batch.allow-latency' = '5s';
  3. SET 'table.exec.mini-batch.size' = '200';
  4. SET 'table.optimizer.agg-phase-strategy' = 'ONE_PHASE';
  6. CREATE TABLE MyTable (
  7.   a BIGINT,
  8.   b INT NOT NULL,
  9.   c VARCHAR,
  10.   d BIGINT
  11. ) WITH (
  12.   'connector' = 'values',
  13.   'bounded' = 'false');
  15. EXPLAIN PLAN_ADVICE
  16. SELECT
  17.   AVG(a) AS avg_a,
  18.   COUNT(*) AS cnt,
  19.   COUNT(b) AS cnt_b,
  20.   MIN(b) AS min_b,
  21.   MAX(c) FILTER (WHERE a > 1) AS max_c
  22. FROM MyTable;

NODE_LEVEL ADVICE

  1. == Optimized Physical Plan With Advice ==
  2. Calc(select=[avg_a, cnt, cnt AS cnt_b, min_b, max_c])
  3. +- GroupAggregate(advice=[1], select=[AVG(a) AS avg_a, COUNT(*) AS cnt, MIN(b) AS min_b, MAX(c) FILTER $f3 AS max_c])
  4.    +- Exchange(distribution=[single])
  5.       +- Calc(select=[a, b, c, IS TRUE(>(a, 1)) AS $f3])
  6.          +- MiniBatchAssigner(interval=[10000ms], mode=[ProcTime])
  7.             +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c], metadata=[]]], fields=[a, b, c])
  9. advice[1]: [ADVICE] You might want to enable local-global two-phase optimization by configuring ('table.optimizer.agg-phase-strategy' to 'AUTO').

若检测到存在 NDU 问题风险时，优化器（optimizer）将会把建议内容附在最后。

SQL2

  1. CREATE TABLE MyTable (
  2.   a INT,
  3.   b BIGINT,
  4.   c STRING,
  5.   d INT,
  6.   `day` AS DATE_FORMAT(CURRENT_TIMESTAMP, 'yyMMdd'),
  7.   PRIMARY KEY (a, c) NOT ENFORCED
  8. ) WITH (
  9.   'connector' = 'values',
  10.   'changelog-mode' = 'I,UA,UB,D'  
  11. );
  13. CREATE TABLE MySink (
  14.  a INT,
  15.  b BIGINT,
  16.  c STRING,
  17.  PRIMARY KEY (a) NOT ENFORCED
  18. ) WITH (
  19.  'connector' = 'values',
  20.  'sink-insert-only' = 'false'
  21. );
  23. EXPLAIN PLAN_ADVICE
  24. INSERT INTO MySink
  25. SELECT a, b, `day`
  26. FROM MyTable
  27. WHERE b > 100;

QUERY_LEVEL WARNING

  1. == Optimized Physical Plan With Advice ==
  2. Sink(table=[default_catalog.default_database.MySink], fields=[a, b, day], upsertMaterialize=[true])
  3. +- Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day], where=[>(b, 100)])
  4.    +- TableSourceScan(table=[[default_catalog, default_database, MyTable, filter=[], project=[a, b], metadata=[]]], fields=[a, b])
  6. advice[1]: [WARNING] The column(s): day(generated by non-deterministic function: CURRENT_TIMESTAMP ) can not satisfy the determinism requirement for correctly processing update message('UB'/'UA'/'D' in changelogMode, not 'I' only), this usually happens when input node has no upsertKey(upsertKeys=[{}]) or current node outputs non-deterministic update messages. Please consider removing these non-deterministic columns or making them deterministic by using deterministic functions.
  8. related rel plan:
  9. Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), _UTF-16LE'yyMMdd') AS day], where=[>(b, 100)], changelogMode=[I,UB,UA,D])
  10. +- TableSourceScan(table=[[default_catalog, default_database, MyTable, filter=[], project=[a, b], metadata=[]]], fields=[a, b], changelogMode=[I,UB,UA,D])

若未检测到问题，优化器（optimizer）将会在计划最后附上 “No available advice” 作为提示。

SQL3

  1. CREATE TABLE MyTable (
  2.   a INT,
  3.   b BIGINT,
  4.   c STRING,
  5.   d INT
  6. ) WITH (
  7.   'connector' = 'values',
  8.   'changelog-mode' = 'I'  
  9. );
  11. EXPLAIN PLAN_ADVICE
  12. SELECT * FROM MyTable WHERE b > 100;

NO_AVAILABLE ADVICE

  1. == Optimized Physical Plan With Advice ==
  2. Calc(select=[a, b, c, d], where=[>(b, 100)])
  3. +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d])
  5. No available advice...

JSON_EXECUTION_PLAN

生成 json 格式的程序执行计划。

语法

  1. EXPLAIN [([ExplainDetail[, ExplainDetail]*]) | PLAN FOR] <query_statement_or_insert_statement_or_statement_set>
  3. statement_set:
  4. EXECUTE STATEMENT SET
  5. BEGIN
  6. insert_statement;
  7. ...
  8. insert_statement;
  9. END;

关于 query 的语法，请查阅 Queries 页面。 关于 INSERT 的语法，请查阅 INSERT 页面。