EXPLAIN Statements

EXPLAIN statements are used to explain the logical and optimized query plans of a query or an INSERT statement.

Run an EXPLAIN statement

Java

EXPLAIN statements can be executed with the executeSql() method of the TableEnvironment. The executeSql() method returns explain result for a successful EXPLAIN operation, otherwise will throw an exception.

The following examples show how to run an EXPLAIN statement in TableEnvironment.

Scala

EXPLAIN statements can be executed with the executeSql() method of the TableEnvironment. The executeSql() method returns explain result for a successful EXPLAIN operation, otherwise will throw an exception.

The following examples show how to run an EXPLAIN statement in TableEnvironment.

Python

EXPLAIN statements can be executed with the execute_sql() method of the TableEnvironment. The execute_sql() method returns explain result for a successful EXPLAIN operation, otherwise will throw an exception.

The following examples show how to run an EXPLAIN statement in TableEnvironment.

SQL CLI

EXPLAIN statements can be executed in SQL CLI.

The following examples show how to run an EXPLAIN statement in SQL CLI.

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  4. // register a table named "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. // explain SELECT statement through TableEnvironment.explainSql()
  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. // explain SELECT statement through TableEnvironment.executeSql()
  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. // register a table named "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. // explain SELECT statement through TableEnvironment.explainSql()
  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. // explain SELECT statement through TableEnvironment.executeSql()
  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. table_env = StreamTableEnvironment.create(...)
  3. t_env.execute_sql("CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  4. t_env.execute_sql("CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH ('connector' = 'datagen')")
  6. # explain SELECT statement through TableEnvironment.explain_sql()
  7. explanation1 = t_env.explain_sql(
  8.     "SELECT `count`, COUNT(word) FROM (" 
  9.     "MyTable1 WHERE word LIKE 'F%' "
  10.     "UNION ALL "
  11.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  12. print(explanation1)
  14. # explain SELECT statement through TableEnvironment.execute_sql()
  15. table_result = t_env.execute_sql(
  16.     "EXPLAIN PLAN FOR "
  17.     "SELECT `count`, COUNT(word) FROM (" 
  18.     "MyTable1 WHERE word LIKE 'F%' "
  19.     "UNION ALL "
  20.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  21. table_result.print()
  23. table_result2 = t_env.execute_sql(
  24.     "EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, PLAN_ADVICE, JSON_EXECUTION_PLAN "
  25.     "SELECT `count`, COUNT(word) FROM (" 
  26.     "MyTable1 WHERE word LIKE 'F%' "
  27.     "UNION ALL "
  28.     "SELECT `count`, word FROM MyTable2) tmp GROUP BY `count`")
  29. 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`;

The EXPLAIN result is:

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 defines the types of details for explain result.

ESTIMATED_COST

Specify ESTIMATED_COST will inform the optimizer to attach the estimated optimal cost of each physical rel node to the output.

  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

Specify CHANGELOG_MODE will inform the optimizer to attach the changelog mode (see more details at Dynamic Tables) of each physical rel node to the ouptput.

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

PLAN_ADVICE

PLAN_ADVICE is supported since Flink version 1.17.

Specify PLAN_ADVICE will inform the optimizer to analyze the optimized physical plan to provide the potential risk warnings and/or optimization advice. Meanwhile, it will change the title of “Optimized Physical Plan” to “Optimized Physical Plan with Advice” as the highlight.

Plan advice is categorized by Kind and Scope.

Advice KindDescription
WARNINGIt reveals potential data correctness risks
ADVICEIt suggests potential SQL optimizer tuning configuration
Advice ScopeDescription
QUERY_LEVELIt provides advice from a global view, targeting the entire query
NODE_LEVELIt provides advice to a specific rel node

Flink SQL provides the plan advice targeting the following issues

If GroupAggregate is detected and can be optimized to the local-global aggregation, the optimizer will tag advice id to the GroupAggregate rel node, and suggests users to update configurations.

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').

If NDU issue is detected, the optimizer will append the warning at the end of the physical plan.

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])

If no warning or advice is detected, the optimizer will append a notice that “No available advice” at the end of the physical plan.

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

Specify JSON_EXECUTION_PLAN will inform the optimizer to attach the json-format execution plan of the program to the output.

Syntax

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

For query syntax, please refer to Queries page. For insert syntax, please refer to INSERT page.