[multistage] enhance physical plan explain

[walterddr]

currently, "physical plan explain" added in #11052 will create a dag-like structure with all the workerIDs printed out
but there are still several improvements we can add:

1. sometimes it only prints the first one due to repetition for example
2. it doesn't have details regarding the logical node (such as project columns etc)

[0]@192.168.1.108:56541 MAIL_RECEIVE(RANDOM_DISTRIBUTED)
├── [1]@192.168.1.108:56595 MAIL_SEND(RANDOM_DISTRIBUTED)->{[0]@192.168.1.108@{56541,56541}|[0]} (Subtree Omitted)
└── [1]@192.168.1.108:56589 MAIL_SEND(RANDOM_DISTRIBUTED)->{[0]@192.168.1.108@{56541,56541}|[0]}
   └── [1]@192.168.1.108:56589 AGGREGATE_FINAL <---- this should really be on 2 servers
      └── [1]@192.168.1.108:56589 MAIL_RECEIVE(HASH_DISTRIBUTED)
         ├── [2]@192.168.1.108:56595 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@192.168.1.108@{56595,56596}|[0],[1]@192.168.1.108@{56589,56590}|[1]} (Subtree Omitted) <---- subtree is omitted b/c they are the same except for the server/worker ID
         └── [2]@192.168.1.108:56589 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@192.168.1.108@{56595,56596}|[0],[1]@192.168.1.108@{56589,56590}|[1]}  <---- mailbox send includes a list of receiving mailbox
            └── [2]@192.168.1.108:56589 AGGREGATE_LEAF <---- this should really be on 2 servers
               └── [2]@192.168.1.108:56589 JOIN <---- this should really be on 2 servers
                  ├── [2]@192.168.1.108:56589 MAIL_RECEIVE(HASH_DISTRIBUTED)
                  │   └── [3]@192.168.1.108:56589 MAIL_SEND(HASH_DISTRIBUTED)->{[2]@192.168.1.108@{56595,56596}|[0],[2]@192.168.1.108@{56589,56590}|[1]}
                  │      └── [3]@192.168.1.108:56589 PROJECT <---- missing project columns
                  │         └── [3]@192.168.1.108:56589 TABLE SCAN (A) null
                  └── [2]@192.168.1.108:56589 MAIL_RECEIVE(HASH_DISTRIBUTED)
                     └── [4]@192.168.1.108:56595 MAIL_SEND(HASH_DISTRIBUTED)->{[2]@192.168.1.108@{56595,56596}|[0],[2]@192.168.1.108@{56589,56590}|[1]}
                        └── [4]@192.168.1.108:56595 PROJECT
                           └── [4]@192.168.1.108:56595 TABLE SCAN (B) null

i would suggest

• all nodes except mailbox send and mailbox receive shouldn't have server/worker info attached. e.g. only the stage/fragment ID
• attach logical info to the nodes as well
• as a side note, do not attach plan in error messages when execution failed. printing it out in the log should be suffice and it will make the error message much simpler comprehend

[abhioncbr]

I can look into it if not urgent.

[walterddr]

thank you @abhioncbr this is not urgent.

[robertzych]

@abhioncbr Are you still looking into this?

[abhioncbr]

@robertzych, I haven't started on this, if yiu want to give a try, please go ahead. Thanks.

[robertzych]

@walterddr Here's a summary of the topics that we'd like to get community feedback on:

1. Should we continue to de-dup the plan by using "Subtree Omitted" or include all sending hosts on the same line? See examples below.

Subtree Omitted:

EXPLAIN IMPLEMENTATION PLAN EXCLUDING ATTRIBUTES FOR SELECT col1, COUNT(*) FROM a GROUP BY col1
[0]@localhost:3 MAIL_RECEIVE(BROADCAST_DISTRIBUTED) [col1(STRING),$f1(LONG)]
├── [1]@localhost:1 MAIL_SEND(BROADCAST_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[0]@localhost@{3,3}|[0]} (Subtree Omitted)
└── [1]@localhost:2 MAIL_SEND(BROADCAST_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[0]@localhost@{3,3}|[0]}
    └── [1]@localhost:2 AGGREGATE_FINAL [col1(STRING),$f1(LONG)]
        └── [1]@localhost:2 MAIL_RECEIVE(HASH_DISTRIBUTED) [col1(STRING),$f1(LONG)]
            ├── [2]@localhost:1 MAIL_SEND(HASH_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[1]@localhost@{1,1}|[1],[1]@localhost@{2,2}|[0]} (Subtree Omitted)
            └── [2]@localhost:2 MAIL_SEND(HASH_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[1]@localhost@{1,1}|[1],[1]@localhost@{2,2}|[0]}
                └── [2]@localhost:2 AGGREGATE_LEAF [col1(STRING),$f1(LONG)]
                    └── [2]@localhost:2 TABLE SCAN (a) [col1(STRING),col2(STRING),col3(INT),col4(BIG_DECIMAL),col5(BOOLEAN),col6(INT),ts(LONG)]

All sending hosts on the same line:

EXPLAIN IMPLEMENTATION PLAN EXCLUDING ATTRIBUTES FOR SELECT col1, COUNT(*) FROM a GROUP BY col1
[0]@localhost:3 MAIL_RECEIVE(BROADCAST_DISTRIBUTED) [col1(STRING),$f1(LONG)]
└── [1]@{localhost:1,localhost:2} MAIL_SEND(BROADCAST_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[0]@localhost@{3,3}|[0]}
    └── [1]@{localhost:1,localhost:2} AGGREGATE_FINAL [col1(STRING),$f1(LONG)]
        └── [1]{localhost:1,localhost:2} MAIL_RECEIVE(HASH_DISTRIBUTED) [col1(STRING),$f1(LONG)]
            └── [2]@{localhost:1,localhost:2} MAIL_SEND(HASH_DISTRIBUTED) [col1(STRING),$f1(LONG)]->{[1]@localhost@{1,1}|[1],[1]@localhost@{2,2}|[0]}
                └── [2]@{localhost:1,localhost:2} AGGREGATE_LEAF [col1(STRING),$f1(LONG)]
                    └── [2]@{localhost:1,localhost:2} TABLE SCAN (a) [col1(STRING),col2(STRING),col3(INT),col4(BIG_DECIMAL),col5(BOOLEAN),col6(INT),ts(LONG)]

2. Should mailbox send/receive pairs share indentation levels?

3. Should we ensure all column names and types are listed in each node (eg [col1(STRING),$f1(LONG)])?

4. Are references like $f1 acceptable, or must they include the original name (eg count)?

[walterddr]

here is just my opinion

1. either approach seems fine to me. subtree omit will be more verbose (print multiple lines); and all sending host on same line will make the line longer; it's really an aesthetic choice

   • however if we ever decided to have different subtree, it will be difficult to print
   • the above is highly unlikely and even if there is, there should be only differences based on physical properties such as one subtree on hostA has 2 segments and hostB has 0 segments and thus being optimized into a empty response.

2. mailbox send/receive should share indentation IMO b/c they technically are one exchange node

3. we should list all column types and reference names

   • but we can also do disable/enable via option, this is not a requirement IMO

4. operand reference is much better b/c it tells the relationship between adjacent nodes (e.g PROJECT[$2(STRING), $5(INT)] tells that the 2nd and 5th columns are read out from previous node's output)

   • however, operator, such as function names, should be parsed out (e.g. $f1 is confusing)

Please definitely share your thoughts @ankitsultana @Jackie-Jiang @xiangfu0