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 = tEnv.explainSql(
  10.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  11.   "UNION ALL " + 
  12.   "SELECT `count`, word FROM MyTable2");
  13. System.out.println(explanation);
  15. // 调用 TableEnvironment.executeSql() 来解释 SELECT 语句
  16. TableResult tableResult = tEnv.executeSql(
  17.   "EXPLAIN PLAN FOR " + 
  18.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  19.   "UNION ALL " + 
  20.   "SELECT `count`, word FROM MyTable2");
  21. tableResult.print();
  23. TableResult tableResult2 = tEnv.executeSql(
  24.   "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, JSON_EXECUTION_PLAN " + 
  25.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  26.   "UNION ALL " + 
  27.   "SELECT `count`, word FROM MyTable2");
  28. 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.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  11.   "UNION ALL " + 
  12.   "SELECT `count`, word FROM MyTable2")
  13. println(explanation)
  15. // 调用 TableEnvironment.executeSql() 来解释 SELECT 语句
  16. val tableResult = tEnv.executeSql(
  17.   "EXPLAIN PLAN FOR " + 
  18.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  19.   "UNION ALL " + 
  20.   "SELECT `count`, word FROM MyTable2")
  21. tableResult.print()
  23. val tableResult2 = tEnv.executeSql(
  24.   "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, JSON_EXECUTION_PLAN " + 
  25.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
  26.   "UNION ALL " + 
  27.   "SELECT `count`, word FROM MyTable2")
  28. 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`, word FROM MyTable1 WHERE word LIKE 'F%' "
  10.     "UNION ALL "
  11.     "SELECT `count`, word FROM MyTable2")
  12. print(explanation1)
  14. # 调用 TableEnvironment.execute_sql() 来解释 SELECT 语句
  15. table_result = t_env.execute_sql(
  16.     "EXPLAIN PLAN FOR "
  17.     "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' "
  18.     "UNION ALL "
  19.     "SELECT `count`, word FROM MyTable2")
  20. table_result.print()
  22. table_result2 = t_env.execute_sql(
  23.     "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, JSON_EXECUTION_PLAN "
  24.     "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' "
  25.     "UNION ALL "
  26.     "SELECT `count`, word FROM MyTable2")
  27. 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`, word FROM MyTable1 WHERE word LIKE 'F%' 
  8. > UNION ALL 
  9. > SELECT `count`, word FROM MyTable2;
  11. Flink SQL> EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, JSON_EXECUTION_PLAN SELECT `count`, word FROM MyTable1 
  12. > WHERE word LIKE 'F%' 
  13. > UNION ALL 
  14. > SELECT `count`, word FROM MyTable2;

EXPLAIN 的结果如下:

EXPLAIN PLAN

  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3. :- LogicalProject(count=[$0], word=[$1])
  4. :  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  5. :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
  6. +- LogicalProject(count=[$0], word=[$1])
  7.    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
  9. == Optimized Physical Plan ==
  10. Union(all=[true], union=[count, word])
  11. :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  12. :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  13. +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])
  15. == Optimized Execution Plan ==
  16. Union(all=[true], union=[count, word])
  17. :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  18. :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  19. +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])

EXPLAIN PLAN WITH DETAILS

  1. == Abstract Syntax Tree ==
  2. LogicalUnion(all=[true])
  3. :- LogicalProject(count=[$0], word=[$1])
  4. :  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
  5. :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
  6. +- LogicalProject(count=[$0], word=[$1])
  7.    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
  9. == Optimized Physical Plan ==
  10. 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}
  11. :- 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}
  12. :  +- 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}
  13. +- 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}
  15. == Optimized Execution Plan ==
  16. Union(all=[true], union=[count, word])
  17. :- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
  18. :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
  19. +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])
  21. == Physical Execution Plan ==
  22. {
  23.   "nodes" : [ {
  24.     "id" : 37,
  25.     "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
  26.     "pact" : "Data Source",
  27.     "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
  28.     "parallelism" : 1
  29.   }, {
  30.     "id" : 38,
  31.     "type" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
  32.     "pact" : "Operator",
  33.     "contents" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
  34.     "parallelism" : 1,
  35.     "predecessors" : [ {
  36.       "id" : 37,
  37.       "ship_strategy" : "FORWARD",
  38.       "side" : "second"
  39.     } ]
  40.   }, {
  41.     "id" : 39,
  42.     "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
  43.     "pact" : "Data Source",
  44.     "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
  45.     "parallelism" : 1
  46.   } ]

ExplainDetails

  1. 使用指定的 explainDetail 类型来打印语句的计划。
  3. ESTIMATED_COST：生成优化器（optimizer）估算的物理节点相关的成本信息, 
  4. 例如：TableSourceScan(..., cumulative cost ={1.0E8 rows, 1.0E8 cpu, 2.4E9 io, 0.0 network, 0.0 memory})
  6. CHANGELOG_MODE：为每个物理 RelNode 生成 changelog mode。
  7. 例如：GroupAggregate(..., changelogMode=[I,UA,D])
  9. 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 页面。