The First Cry of Atom

Visualize Execution Plan in Presto

In general, visualization is an essential technique to understand what is happening. The software does not always provide informative metrics to us for debugging and inspection. We must get them visualized proactively. Notably, it is hard to investigate how a distributed program works without well-defined visualization tools due to the nature of its asynchronous and uncertainty. In this article, I’m going to demonstrate how to visualize the execution plan of Presto which is one of the most advanced distributed execution systems.

Environment

You can quickly try to run the following code by using docker-presto-cluster.

$ make run

The Presto cluster with version 318 is launched in your local machine. Please make sure to install the Presto CLI to run the query to the cluster. The instruction is here.

The following code connects to the local coordinator with specifying tpch catalog and tiny schema as default.

$ ./presto-cli-318-SNAPSHOT-executable.jar --server localhost:8080 --catalog tpch --schema tiny

First, let’s look at how to print the logical plan.

Logical Plan

EXPLAIN is a significant feature to print the logical plan used in various kind of implementations supporting SQL. Presto also shows the logical plan as default by using EXPLAIN. EXPLAIN (TYPE LOGICAL) does the same thing.

presto:tiny> explain select custkey, cnt from (select custkey, count(1) cnt from customer group by custkey) where cnt > 10;
                                                    Query Plan
------------------------------------------------------------------------------------------------------------------
 Output[custkey, cnt]
 │   Layout: [custkey:bigint, count:bigint]
 │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
 │   cnt := count
 └─ RemoteExchange[GATHER]
    │   Layout: [custkey:bigint, count:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    └─ Filter[filterPredicate = ("count" > BIGINT '10')]
       │   Layout: [custkey:bigint, count:bigint]
       │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
       └─ Aggregate(FINAL)[custkey]
          │   Layout: [custkey:bigint, count:bigint]
          │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
          │   count := count("count_16")
          └─ LocalExchange[HASH][$hashvalue] ("custkey")
             │   Layout: [custkey:bigint, count_16:bigint, $hashvalue:bigint]
             │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
             └─ RemoteExchange[REPARTITION][$hashvalue_17]
                │   Layout: [custkey:bigint, count_16:bigint, $hashvalue_17:bigint]
                │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
                └─ Project[]
                   │   Layout: [custkey:bigint, count_16:bigint, $hashvalue_18:bigint]
                   │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
                   │   $hashvalue_18 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("custkey"), 0))
                   └─ Aggregate(PARTIAL)[custkey]
                      │   Layout: [custkey:bigint, count_16:bigint]
                      │   count_16 := count(*)
                      └─ TableScan[tpch:customer:sf0.01]
                             Layout: [custkey:bigint]
                             Estimates: {rows: 1500 (13.18kB), cpu: 13.18k, memory: 0B, network: 0B}
                             custkey := tpch:custkey

It prints the hierarchy of logical operations of the query.

Distributed Plan

To print the physical plan, which is a real execution plan of the distributed environment, you can specify the type DISTRIBUTED. A fragment represents a stage of the distributed plan. Presto scheduler schedules the execution by each stage, and stages can be run on separated instances.

presto:tiny> explain (type distributed) select custkey, cnt from (select custkey, count(1) cnt from customer group by custkey) where cnt > 10;
                                             Query Plan
----------------------------------------------------------------------------------------------------
 Fragment 0 [SINGLE]
     Output layout: [custkey, count]
     Output partitioning: SINGLE []
     Stage Execution Strategy: UNGROUPED_EXECUTION
     Output[custkey, cnt]
     │   Layout: [custkey:bigint, count:bigint]
     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
     │   cnt := count
     └─ RemoteSource[1]
            Layout: [custkey:bigint, count:bigint]

 Fragment 1 [HASH]
     Output layout: [custkey, count]
     Output partitioning: SINGLE []
     Stage Execution Strategy: UNGROUPED_EXECUTION
     Filter[filterPredicate = ("count" > BIGINT '10')]
     │   Layout: [custkey:bigint, count:bigint]
     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
     └─ Aggregate(FINAL)[custkey]
        │   Layout: [custkey:bigint, count:bigint]
        │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
        │   count := count("count_16")
        └─ LocalExchange[HASH][$hashvalue] ("custkey")
           │   Layout: [custkey:bigint, count_16:bigint, $hashvalue:bigint]
           │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
           └─ RemoteSource[2]
                  Layout: [custkey:bigint, count_16:bigint, $hashvalue_17:bigint]

 Fragment 2 [SOURCE]
     Output layout: [custkey, count_16, $hashvalue_18]
     Output partitioning: HASH [custkey][$hashvalue_18]
     Stage Execution Strategy: UNGROUPED_EXECUTION
     Project[]
     │   Layout: [custkey:bigint, count_16:bigint, $hashvalue_18:bigint]
     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
     │   $hashvalue_18 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("custkey"), 0))
     └─ Aggregate(PARTIAL)[custkey]
        │   Layout: [custkey:bigint, count_16:bigint]
        │   count_16 := count(*)
        └─ TableScan[tpch:customer:sf0.01, grouped = false]
               Layout: [custkey:bigint]
               Estimates: {rows: 1500 (13.18kB), cpu: 13.18k, memory: 0B, network: 0B}
               custkey := tpch:custkey

IO

We may sometimes want to focus on the IO of the query. Like which table is the query reading? TYPE IO brings us the information around the table and schemas the query reads in JSON format.

presto:tiny> explain (type io) select custkey, cnt from (select custkey, count(1) cnt from customer group by custkey) where cnt > 10;
           Query Plan
---------------------------------
 {
   "inputTableColumnInfos" : [ {
     "table" : {
       "catalog" : "tpch",
       "schemaTable" : {
         "schema" : "sf0.01",
         "table" : "customer"
       }
     },
     "columnConstraints" : [ ]
   } ]
 }
(1 row)

Graphviz

The printed information in the console is basically sufficiently useful, but we want to make it better in terms of the visibility. You can use the format option to print in the format compatible with Graphviz.

presto:tiny> explain (format graphviz) select custkey, cnt from (select custkey, count(1) cnt from customer group by custkey) where cnt > 10;
                                                                                         Query Plan
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 digraph logical_plan {
 subgraph cluster_graphviz_plan {
 label = "SINGLE"
 plannode_1[label="{Output[custkey, cnt]}", style="rounded, filled", shape=record, fillcolor=white];
 plannode_2[label="{ExchangeNode[GATHER]|\"custkey\", \"count\"}", style="rounded, filled", shape=record, fillcolor=gold];
 plannode_3[label="{Filter|(\"count\" \> BIGINT '10')}", style="rounded, filled", shape=record, fillcolor=yellow];
 plannode_4[label="{Aggregate[FINAL]|count := count(\"count_16\")\n}", style="rounded, filled", shape=record, fillcolor=chartreuse3];
 plannode_5[label="{ExchangeNode[REPARTITION]|\"custkey\"}", style="rounded, filled", shape=record, fillcolor=gold];
 plannode_6[label="{ExchangeNode[REPARTITION]|\"custkey\"}", style="rounded, filled", shape=record, fillcolor=gold];
 plannode_7[label="{Project|$hashvalue_18 := \"combine_hash\"(bigint '0', COALESCE(\"$operator$hash_code\"(\"custkey\"), 0))\n}", style="rounded, filled", shape=record, fillcolor=bisque];
 plannode_8[label="{Aggregate[PARTIAL]|count_16 := count(*)\n}", style="rounded, filled", shape=record, fillcolor=chartreuse3];
 plannode_9[label="{TableScan[tpch:customer:sf0.01]}", style="rounded, filled", shape=record, fillcolor=deepskyblue];
 }
 plannode_1 -> plannode_2;
 plannode_2 -> plannode_3;
 plannode_3 -> plannode_4;
 plannode_4 -> plannode_5;
 plannode_5 -> plannode_6;
 plannode_6 -> plannode_7;
 plannode_7 -> plannode_8;
 plannode_8 -> plannode_9;
 }

(1 row)

Of course, it is not so useful as it is. Copy and paste it in the file (plan.dot). You can use the dot tool to convert the text format in the image. Please see here about the installation.

$ dot -Tpng plan.dot > plan.png

This is the graph generated by dot. Although it is a little cumbersome to create the image file like this, it is more informative and intuitive to grasp the overview of the dependencies between each operator.

plan

You can use Graphviz format for distributed plan too. It’s fun to see how these two plans can be different from each other. Let’s try to make use of the tool to pursue more performant distributed queries.

Thanks!

Image by Andrew Martin from Pixabay