EXPLAIN Statements

    Java

    EXPLAIN statements can be executed with the 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.

    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. 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.   "SELECT `count`, word FROM MyTable1 WHERE word LIKE 'F%' " +
    11.   "UNION ALL " + 
    12.   "SELECT `count`, word FROM MyTable2")
    13. println(explanation)
    15. // explain SELECT statement through TableEnvironment.executeSql()
    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()
    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")

    Python

    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;

    The EXPLAIN result is:

    EXPLAIN PLAN

    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. :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
    18. +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])
    20. == Physical Execution Plan ==
    21. {
    22.   "nodes" : [ {
    23.     "id" : 37,
    24.     "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
    25.     "pact" : "Data Source",
    26.     "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
    27.     "parallelism" : 1
    28.   }, {
    29.     "id" : 38,
    30.     "type" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
    31.     "pact" : "Operator",
    32.     "contents" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
    33.     "parallelism" : 1,
    34.     "predecessors" : [ {
    35.       "id" : 37,
    36.       "ship_strategy" : "FORWARD",
    37.       "side" : "second"
    38.     } ]
    39.   }, {
    40.     "id" : 39,
    41.     "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
    42.     "pact" : "Data Source",
    43.     "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
    44.     "parallelism" : 1
    45.   } ]
    1. EXPLAIN [([ExplainDetail[, ExplainDetail]*]) | PLAN FOR] <query_statement_or_insert_statement>

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