Hive on TezのEXPLAINを読み解く

#Hadoop

じょんすみす

2017.05.25

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、小澤です。

RDBではおなじみのEXPLAIN、実行計画を確認しクエリを最適化したり、インデックスの貼り方を考えたりするのによく使われるかと思います。このEXPLAINですが、Hiveでも利用可能です。 HiveのEXPLAINはMapReduceやTez, Sparkのジョブに変換されるため、少々見方が特殊でこれらの知識も必要になります。

実行エンジンによって表示され方が異なるため、今回はHive on Tezに限定してこのEXPLAINの見方を解説していきます。

今回利用するクエリ

今回は、TPC-DSというOLAPなどの分析系クエリのベンチマークで利用されるデータを利用して見ていきたいと思います。 TPC-DSの詳細については解説しませんので、気になる方は公式の情報をご確認ください。

TPC-DS

HiveでTPC-DSを利用できる環境を整えるために、以下のスクリプトを利用します。

hortonworks/hive-testbench

この中のsample-queries-tpcdsというディレクトリを見ていただくと、サンプルのクエリがかなりの数あることがわかります。今回はこの中のいくつかの実行計画を見てみます。

実際に見てみる

では、いくつかのクエリで実行計画を見てみましょう

query12.sql

１つ目は、ファイル名がquery12.sqlとなっているものです。

クエリの内容は以下のようになります。

select  i_item_desc 
      ,i_category 
      ,i_class 
      ,i_current_price
      ,i_item_id
      ,sum(ws_ext_sales_price) as itemrevenue 
      ,sum(ws_ext_sales_price)*100/sum(sum(ws_ext_sales_price)) over
          (partition by i_class) as revenueratio
from	
	web_sales
    	,item 
    	,date_dim
where 
	web_sales.ws_item_sk = item.i_item_sk 
  	and item.i_category in ('Jewelry', 'Sports', 'Books')
  	and web_sales.ws_sold_date_sk = date_dim.d_date_sk
	and date_dim.d_date between '2001-01-12' and '2001-02-11'
group by 
	i_item_id
        ,i_item_desc 
        ,i_category
        ,i_class
        ,i_current_price
order by 
	i_category
        ,i_class
        ,i_item_id
        ,i_item_desc
        ,revenueratio
limit 100;

結果は以下のようになります。

OK
Plan optimized by CBO.

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 7 (SIMPLE_EDGE)
Reducer 3 <- Map 8 (SIMPLE_EDGE), Reducer 2 (SIMPLE_EDGE)
Reducer 4 <- Reducer 3 (SIMPLE_EDGE)
Reducer 5 <- Reducer 4 (SIMPLE_EDGE)
Reducer 6 <- Reducer 5 (SIMPLE_EDGE)

Stage-0
  Fetch Operator
    limit:100
    Stage-1
      Reducer 6
      File Output Operator [FS_28]
        Limit [LIM_27] (rows=100 width=259)
          Number of rows:100
          Select Operator [SEL_26] (rows=4353445 width=259)
            Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6"]
          <-Reducer 5 [SIMPLE_EDGE]
            SHUFFLE [RS_25]
              Select Operator [SEL_23] (rows=4353445 width=259)
                Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6"]
                PTF Operator [PTF_22] (rows=4353445 width=259)
                  Function definitions:[{},{"name:":"windowingtablefunction","order by:":"_col3 ASC NULLS FIRST","partition by:":"_col3"}]
                  Select Operator [SEL_21] (rows=4353445 width=259)
                    Output:["_col0","_col1","_col2","_col3","_col4","_col5"]
                  <-Reducer 4 [SIMPLE_EDGE]
                    SHUFFLE [RS_20]
                      PartitionCols:_col3
                      Select Operator [SEL_19] (rows=4353445 width=259)
                        Output:["_col0","_col1","_col2","_col3","_col4","_col5"]
                        Group By Operator [GBY_18] (rows=4353445 width=259)
                          Output:["_col0","_col1","_col2","_col3","_col4","_col5"],aggregations:["sum(VALUE._col0)"],keys:KEY._col0, KEY._col1, KEY._col2, KEY._col3, KEY._col4
                        <-Reducer 3 [SIMPLE_EDGE]
                          SHUFFLE [RS_17]
                            PartitionCols:_col0, _col1, _col2, _col3, _col4
                            Group By Operator [GBY_16] (rows=8706890 width=259)
                              Output:["_col0","_col1","_col2","_col3","_col4","_col5"],aggregations:["sum(_col1)"],keys:_col8, _col7, _col4, _col5, _col6
                              Select Operator [SEL_15] (rows=8706890 width=259)
                                Output:["_col8","_col7","_col4","_col5","_col6","_col1"]
                                Merge Join Operator [MERGEJOIN_41] (rows=8706890 width=259)
                                  Conds:RS_12._col2=RS_13._col0(Inner),Output:["_col1","_col4","_col5","_col6","_col7","_col8"]
                                <-Map 8 [SIMPLE_EDGE]
                                  SHUFFLE [RS_13]
                                    PartitionCols:_col0
                                    Select Operator [SEL_8] (rows=36524 width=1123)
                                      Output:["_col0"]
                                      Filter Operator [FIL_36] (rows=36524 width=1123)
                                        predicate:(d_date BETWEEN '2001-01-12' AND '2001-02-11' and d_date_sk is not null)
                                        TableScan [TS_6] (rows=73049 width=1123)
                                          tpcds_bin_partitioned_orc_10@date_dim,date_dim,Tbl:COMPLETE,Col:NONE,Output:["d_date_sk","d_date"]
                                  Dynamic Partitioning Event Operator [EVENT_39] (rows=36524 width=1123)
                                    Group By Operator [GBY_38] (rows=36524 width=1123)
                                      Output:["_col0"],keys:_col0
                                      Select Operator [SEL_37] (rows=36524 width=1123)
                                        Output:["_col0"]
                                         Please refer to the previous Select Operator [SEL_8]
                                <-Reducer 2 [SIMPLE_EDGE]
                                  SHUFFLE [RS_12]
                                    PartitionCols:_col2
                                    Merge Join Operator [MERGEJOIN_40] (rows=7915355 width=259)
                                      Conds:RS_9._col0=RS_10._col0(Inner),Output:["_col1","_col2","_col4","_col5","_col6","_col7","_col8"]
                                    <-Map 1 [SIMPLE_EDGE]
                                      SHUFFLE [RS_9]
                                        PartitionCols:_col0
                                        Select Operator [SEL_2] (rows=7195778 width=259)
                                          Output:["_col0","_col1","_col2"]
                                          Filter Operator [FIL_34] (rows=7195778 width=259)
                                            predicate:ws_item_sk is not null
                                            TableScan [TS_0] (rows=7195778 width=259)
                                              tpcds_bin_partitioned_orc_10@web_sales,web_sales,Tbl:COMPLETE,Col:NONE,Output:["ws_item_sk","ws_ext_sales_price"]
                                    <-Map 7 [SIMPLE_EDGE]
                                      SHUFFLE [RS_10]
                                        PartitionCols:_col0
                                        Select Operator [SEL_5] (rows=51000 width=1445)
                                          Output:["_col0","_col1","_col2","_col3","_col4","_col5"]
                                          Filter Operator [FIL_35] (rows=51000 width=1445)
                                            predicate:((i_category) IN ('Jewelry', 'Sports', 'Books') and i_item_sk is not null)
                                            TableScan [TS_3] (rows=102000 width=1445)
                                              tpcds_bin_partitioned_orc_10@item,item,Tbl:COMPLETE,Col:NONE,Output:["i_item_sk","i_item_id","i_item_desc","i_current_price","i_class","i_category"]

全体としては非常に長い結果が出力されているので、要点をかいつまんで解説していきます。

まず、一番最初の

Plan optimized by CBO.

ですが、これはCBO(Cost Based Optimaizer)という機能が動作していることを意味します。 CBOが使えない場合出力されません。

つぎにどのようなTezのDAGが生成されるかの部分です。 DAGに関してはジョブを実行すると、TezのWebUIからビジュアライズされたものが確認できますが、実行計画のこの部分から同じものを作成することも可能です。

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 7 (SIMPLE_EDGE)
Reducer 3 <- Map 8 (SIMPLE_EDGE), Reducer 2 (SIMPLE_EDGE)
Reducer 4 <- Reducer 3 (SIMPLE_EDGE)
Reducer 5 <- Reducer 4 (SIMPLE_EDGE)
Reducer 6 <- Reducer 5 (SIMPLE_EDGE)

矢印の方向が次の処理になるので、実際のDAGは以下のようになります。

dag12

最後にインデントが入っている長い部分が、実際のDAG上での個々のノードの処理になります。 SQLが内側のサブクエリから順に見ていったほうが読みやすいのと同様、この部分の始点となるもっともインデントの深い部分から順に見ていきます。

まずは、Map1から見てみましょう

<-Map 1 [SIMPLE_EDGE]
  SHUFFLE [RS_9]
    PartitionCols:_col0
    Select Operator [SEL_2] (rows=7195778 width=259)
      Output:["_col0","_col1","_col2"]
      Filter Operator [FIL_34] (rows=7195778 width=259)
        predicate:ws_item_sk is not null
          TableScan [TS_0] (rows=7195778 width=259)
            tpcds_bin_partitioned_orc_10@web_sales,web_sales,Tbl:COMPLETE,Col:NONE,Output:["ws_item_sk","ws_ext_sales_price"]

TableScanでweb_salesというテーブルのデータを読み込んで、 Filter Operatorで必要な行だけに絞り込み、 Select Operatorで必要な列だけに絞り込む。

という流れになっています。

Hiveは、入力されたHiveQLを字句解析・構文解析したのち、それぞれの処理をOperatorという、それぞれに対応した実際の処理に置き換えます。

最後にSHUFFLEでネットワークをまたいだデータの分散が必要になるまでがDAG上のそのノードでの１つの処理となります。

個々のOperatorが具体的にどのような処理を行っているかは、その下にインデントされたかたちで記載されています。 Map1のTableScan場合、web_salesテーブルを読み込んでおり、必要なカラムはws_items_skとws_ext_sales_priceとなっています。 SQLの方と照らし合わせると、この2カラムがSELECT句とWHERE句で実際に利用されているのがわかるかと思います。

また、実際にはws_sold_date_skも利用されているように見えますが、こちらはparitionキーとなっているため、実際のデータ内には含まれていません。

Map7なんかも同じような感じでみればだいたい何をしているか確認できるでしょう。

Reduce2を見て見ましょう

<-Reducer 2 [SIMPLE_EDGE]
  SHUFFLE [RS_12]
    PartitionCols:_col2
      Merge Join Operator [MERGEJOIN_40] (rows=7915355 width=259)
        Conds:RS_9._col0=RS_10._col0(Inner),Output:["_col1","_col2","_col4","_col5","_col6","_col7","_col8"]

こちらはMerge Join Operatorというものが出てきます。これは、JOINを実現する仕組みになっています。 Hadoop上でjoin相当の処理を実現するための方法は複数あり、それぞれにどういった時に利用すると効果的かが異なってきます。 Hiveにも数種類のJoin Operatorが存在しており、そのうちの1つになります。

Hadoop上でのjoinの方法は実行速度の観点から重要な要素になるため、ここでは代表的なもの2つを補足として紹介します。

１つ目は、joinの対象となる2つデータそれぞれが、joinのキーとなる値をMapのkey, 出力対象となるデータをvalueとして出力することで、同一Reduceに結合対象となる2つのテーブルの同じキーを持つデータを渡してやることができます。

join1

この方法が、今回のMerge Join Operatorで行っているものになります。

一方、この後別なクエリで登場するMap Join Operatorで行っている方法はこれとは異なります。 Map Joinでは、joinするときの片方のテーブルを全てMapのメモリ上にのせてしまいます。

こうすることでMap側で各行のデータに対して、結合を行った後の結果を後続の処理に渡してやることができます。

join2

Map JoinはDAG上の単一のノードで処理が完結するため、Merge Joinよりも処理速度の面で効率的です。しかし、片方のテーブルがメモリ上に乗り切るくらいに小さくないと実現できません。

このクエリの最後にMap8を見てみましょう

<-Map 8 [SIMPLE_EDGE]
  SHUFFLE [RS_13]
    PartitionCols:_col0
      Select Operator [SEL_8] (rows=36524 width=1123)
        Output:["_col0"]
        Filter Operator [FIL_36] (rows=36524 width=1123)
          predicate:(d_date BETWEEN '2001-01-12' AND '2001-02-11' and d_date_sk is not null)
            TableScan [TS_6] (rows=73049 width=1123)
              tpcds_bin_partitioned_orc_10@date_dim,date_dim,Tbl:COMPLETE,Col:NONE,Output:["d_date_sk","d_date"]
  Dynamic Partitioning Event Operator [EVENT_39] (rows=36524 width=1123)
    Group By Operator [GBY_38] (rows=36524 width=1123)
      Output:["_col0"],keys:_col0
      Select Operator [SEL_37] (rows=36524 width=1123)
        Output:["_col0"]
          Please refer to the previous Select Operator [SEL_8]

Dynamic Partitioning Event Operatorというものが含まれています。難しい言葉が出てきましたねー。

この部分、実は私もあまり見たことがありませんでした(今までhive.tez.dynamic.partition.pruning=trueにしてなかったのかな？)

この機能が実装されたHIVE-7826のDescriptionを見ると、 Tezの機能を利用して、不要なパーティションを実行時に動的に取り除く機能のようです(間違っていたらご指摘ください。。)

query21.sql

query21はサブクエリを含むものになっています。

select  *
 from(select w_warehouse_name
            ,i_item_id
            ,sum(case when (cast(d_date as date) < cast ('1998-04-08' as date))
	                then inv_quantity_on_hand 
                      else 0 end) as inv_before
            ,sum(case when (cast(d_date as date) >= cast ('1998-04-08' as date))
                      then inv_quantity_on_hand 
                      else 0 end) as inv_after
   from inventory
       ,warehouse
       ,item
       ,date_dim
   where i_current_price between 0.99 and 1.49
     and item.i_item_sk          = inventory.inv_item_sk
     and inventory.inv_warehouse_sk   = warehouse.w_warehouse_sk
     and inventory.inv_date_sk    = date_dim.d_date_sk
     and d_date between '1998-03-09' and '1998-05-07'
   group by w_warehouse_name, i_item_id) x
 where (case when inv_before > 0 
             then inv_after / inv_before 
             else null
             end) between 2.0/3.0 and 3.0/2.0
 order by w_warehouse_name
         ,i_item_id
 limit 100;

OK
Plan optimized by CBO.

Vertex dependency in root stage
Map 1 <- Map 6 (BROADCAST_EDGE)
Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 7 (SIMPLE_EDGE)
Reducer 3 <- Map 8 (SIMPLE_EDGE), Reducer 2 (SIMPLE_EDGE)
Reducer 4 <- Reducer 3 (SIMPLE_EDGE)
Reducer 5 <- Reducer 4 (SIMPLE_EDGE)

Stage-0
  Fetch Operator
    limit:100
    Stage-1
      Reducer 5
      File Output Operator [FS_31]
        Limit [LIM_30] (rows=100 width=19)
          Number of rows:100
          Select Operator [SEL_29] (rows=44292355 width=19)
            Output:["_col0","_col1","_col2","_col3"]
          <-Reducer 4 [SIMPLE_EDGE]
            SHUFFLE [RS_28]
              Filter Operator [FIL_26] (rows=44292355 width=19)
                predicate:CASE WHEN ((_col2 > 0)) THEN ((UDFToDouble(_col3) / UDFToDouble(_col2)) BETWEEN 0.6666666666666666 AND 1.5) ELSE (null) END
                Group By Operator [GBY_25] (rows=88584710 width=19)
                  Output:["_col0","_col1","_col2","_col3"],aggregations:["sum(VALUE._col0)","sum(VALUE._col1)"],keys:KEY._col0, KEY._col1
                <-Reducer 3 [SIMPLE_EDGE]
                  SHUFFLE [RS_24]
                    PartitionCols:_col0, _col1
                    Group By Operator [GBY_23] (rows=177169420 width=19)
                      Output:["_col0","_col1","_col2","_col3"],aggregations:["sum(_col2)","sum(_col3)"],keys:_col0, _col1
                      Select Operator [SEL_21] (rows=177169420 width=19)
                        Output:["_col0","_col1","_col2","_col3"]
                        Merge Join Operator [MERGEJOIN_50] (rows=177169420 width=19)
                          Conds:RS_18._col3=RS_19._col0(Inner),Output:["_col2","_col5","_col7","_col10"]
                        <-Map 8 [SIMPLE_EDGE]
                          SHUFFLE [RS_19]
                            PartitionCols:_col0
                            Select Operator [SEL_11] (rows=36524 width=1123)
                              Output:["_col0","_col1"]
                              Filter Operator [FIL_44] (rows=36524 width=1123)
                                predicate:(d_date BETWEEN '1998-03-09' AND '1998-05-07' and d_date_sk is not null)
                                TableScan [TS_9] (rows=73049 width=1123)
                                  tpcds_bin_partitioned_orc_10@date_dim,date_dim,Tbl:COMPLETE,Col:NONE,Output:["d_date_sk","d_date"]
                          Dynamic Partitioning Event Operator [EVENT_47] (rows=36524 width=1123)
                            Group By Operator [GBY_46] (rows=36524 width=1123)
                              Output:["_col0"],keys:_col0
                              Select Operator [SEL_45] (rows=36524 width=1123)
                                Output:["_col0"]
                                 Please refer to the previous Select Operator [SEL_11]
                        <-Reducer 2 [SIMPLE_EDGE]
                          SHUFFLE [RS_18]
                            PartitionCols:_col3
                            Merge Join Operator [MERGEJOIN_49] (rows=161063106 width=19)
                              Conds:RS_15._col0=RS_16._col0(Inner),Output:["_col2","_col3","_col5","_col7"]
                            <-Map 1 [SIMPLE_EDGE]
                              SHUFFLE [RS_15]
                                PartitionCols:_col0
                                Map Join Operator [MAPJOIN_48] (rows=146421003 width=19)
                                  Conds:SEL_2._col1=RS_13._col0(Inner),HybridGraceHashJoin:true,Output:["_col0","_col2","_col3","_col5"]
                                <-Map 6 [BROADCAST_EDGE]
                                  BROADCAST [RS_13]
                                    PartitionCols:_col0
                                    Select Operator [SEL_5] (rows=10 width=1033)
                                      Output:["_col0","_col1"]
                                      Filter Operator [FIL_42] (rows=10 width=1033)
                                        predicate:w_warehouse_sk is not null
                                        TableScan [TS_3] (rows=10 width=1033)
                                          tpcds_bin_partitioned_orc_10@warehouse,warehouse,Tbl:COMPLETE,Col:NONE,Output:["w_warehouse_sk","w_warehouse_name"]
                                <-Select Operator [SEL_2] (rows=133110000 width=19)
                                    Output:["_col0","_col1","_col2","_col3"]
                                    Filter Operator [FIL_41] (rows=133110000 width=19)
                                      predicate:(inv_warehouse_sk is not null and inv_item_sk is not null)
                                      TableScan [TS_0] (rows=133110000 width=19)
                                        tpcds_bin_partitioned_orc_10@inventory,inventory,Tbl:COMPLETE,Col:NONE,Output:["inv_item_sk","inv_warehouse_sk","inv_quantity_on_hand"]
                            <-Map 7 [SIMPLE_EDGE]
                              SHUFFLE [RS_16]
                                PartitionCols:_col0
                                Select Operator [SEL_8] (rows=51000 width=1445)
                                  Output:["_col0","_col1"]
                                  Filter Operator [FIL_43] (rows=51000 width=1445)
                                    predicate:(i_current_price BETWEEN 0.99 AND 1.49 and i_item_sk is not null)
                                    TableScan [TS_6] (rows=102000 width=1445)
                                      tpcds_bin_partitioned_orc_10@item,item,Tbl:COMPLETE,Col:NONE,Output:["i_item_sk","i_item_id","i_current_price"]

こちらも非常に長い実行計画が表示されていますが、先ほどのクエリになかった点としては、

Map 1 <- Map 6 (BROADCAST_EDGE)

でしょう。この部分の実際の処理は以下のようになっています。

<-Map 1 [SIMPLE_EDGE]
  SHUFFLE [RS_15]
    PartitionCols:_col0
    Map Join Operator [MAPJOIN_48] (rows=146421003 width=19)
      Conds:SEL_2._col1=RS_13._col0(Inner),HybridGraceHashJoin:true,Output:["_col0","_col2","_col3","_col5"]
      <-Map 6 [BROADCAST_EDGE]
        BROADCAST [RS_13]
          PartitionCols:_col0
            Select Operator [SEL_5] (rows=10 width=1033)
              Output:["_col0","_col1"]
                Filter Operator [FIL_42] (rows=10 width=1033)
                  predicate:w_warehouse_sk is not null
                  TableScan [TS_3] (rows=10 width=1033)
                    tpcds_bin_partitioned_orc_10@warehouse,warehouse,Tbl:COMPLETE,Col:NONE,Output:["w_warehouse_sk","w_warehouse_name"]
      <-Select Operator [SEL_2] (rows=133110000 width=19)
        Output:["_col0","_col1","_col2","_col3"]
          Filter Operator [FIL_41] (rows=133110000 width=19)
            predicate:(inv_warehouse_sk is not null and inv_item_sk is not null)
            TableScan [TS_0] (rows=133110000 width=19)
              tpcds_bin_partitioned_orc_10@inventory,inventory,Tbl:COMPLETE,Col:NONE,Output:["inv_item_sk","inv_warehouse_sk","inv_quantity_on_hand"]

これが、先ほどjoinの補足であげたMap Joinの部分になります。 Map6でjoinの対象となるデータを生成したのち、「BROADCAST_EDGE」なので、Map1の全ノードに同じデータを配布しています。 Map1ではメモリ上に保持したそのデータと[SEL_2]までで取得したデータとの結合を行うというわけです。

query32.sql

最後に練習問題(?)としてquery32.sqlを掲載しておきます。こちらのクエリ及び実行計画に関しての詳細は解説しませんが、ここまで内容ですでに読めるようになっているかと思います。

ELECT sum(cs1.cs_ext_discount_amt) as excess_discount_amount
FROM (SELECT cs.cs_item_sk as cs_item_sk,
                             cs.cs_ext_discount_amt as cs_ext_discount_amt
             FROM catalog_sales cs
             JOIN date_dim d ON (d.d_date_sk = cs.cs_sold_date_sk)
             WHERE d.d_date between '2000-01-27' and '2000-04-27') cs1
JOIN item i ON (i.i_item_sk = cs1.cs_item_sk)
JOIN (SELECT cs2.cs_item_sk as cs_item_sk,
                          1.3 * avg(cs_ext_discount_amt) as avg_cs_ext_discount_amt
           FROM (SELECT cs.cs_item_sk as cs_item_sk,
                                        cs.cs_ext_discount_amt as cs_ext_discount_amt
                        FROM catalog_sales cs
                        JOIN date_dim d ON (d.d_date_sk = cs.cs_sold_date_sk)
                        WHERE d.d_date between '2000-01-27' and '2000-04-27') cs2
                        GROUP BY cs2.cs_item_sk) tmp1
ON (i.i_item_sk = tmp1.cs_item_sk)
WHERE i.i_manufact_id = 436 and
               cs1.cs_ext_discount_amt > tmp1.avg_cs_ext_discount_amt;

OK
Plan optimized by CBO.

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 5 (SIMPLE_EDGE)
Reducer 3 <- Map 6 (SIMPLE_EDGE), Reducer 2 (SIMPLE_EDGE), Reducer 9 (SIMPLE_EDGE)
Reducer 4 <- Reducer 3 (SIMPLE_EDGE)
Reducer 8 <- Map 10 (SIMPLE_EDGE), Map 7 (SIMPLE_EDGE)
Reducer 9 <- Reducer 8 (SIMPLE_EDGE)

Stage-0
  Fetch Operator
    limit:-1
    Stage-1
      Reducer 4
      File Output Operator [FS_37]
        Group By Operator [GBY_35] (rows=1 width=8)
          Output:["_col0"],aggregations:["sum(VALUE._col0)"]
        <-Reducer 3 [SIMPLE_EDGE]
          SHUFFLE [RS_34]
            Group By Operator [GBY_33] (rows=1 width=8)
              Output:["_col0"],aggregations:["sum(_col1)"]
              Select Operator [SEL_32] (rows=11558959 width=259)
                Output:["_col1"]
                Filter Operator [FIL_31] (rows=11558959 width=259)
                  predicate:(_col1 > _col5)
                  Merge Join Operator [MERGEJOIN_65] (rows=34676878 width=259)
                    Conds:RS_27._col0=RS_28._col0(Inner),RS_28._col0=RS_29._col0(Inner),Output:["_col1","_col5"]
                  <-Map 6 [SIMPLE_EDGE]
                    SHUFFLE [RS_28]
                      PartitionCols:_col0
                      Select Operator [SEL_12] (rows=51000 width=1445)
                        Output:["_col0"]
                        Filter Operator [FIL_54] (rows=51000 width=1445)
                          predicate:((i_manufact_id = 436) and i_item_sk is not null)
                          TableScan [TS_10] (rows=102000 width=1445)
                            tpcds_bin_partitioned_orc_10@item,i,Tbl:COMPLETE,Col:NONE,Output:["i_item_sk","i_manufact_id"]
                  <-Reducer 2 [SIMPLE_EDGE]
                    SHUFFLE [RS_27]
                      PartitionCols:_col0
                      Merge Join Operator [MERGEJOIN_63] (rows=15762217 width=259)
                        Conds:RS_6._col2=RS_7._col0(Inner),Output:["_col0","_col1"]
                      <-Map 1 [SIMPLE_EDGE]
                        SHUFFLE [RS_6]
                          PartitionCols:_col2
                          Select Operator [SEL_2] (rows=14329288 width=259)
                            Output:["_col0","_col1","_col2"]
                            Filter Operator [FIL_52] (rows=14329288 width=259)
                              predicate:cs_item_sk is not null
                              TableScan [TS_0] (rows=14329288 width=259)
                                tpcds_bin_partitioned_orc_10@catalog_sales,cs,Tbl:COMPLETE,Col:NONE,Output:["cs_item_sk","cs_ext_discount_amt"]
                      <-Map 5 [SIMPLE_EDGE]
                        SHUFFLE [RS_7]
                          PartitionCols:_col0
                          Select Operator [SEL_5] (rows=36524 width=1123)
                            Output:["_col0"]
                            Filter Operator [FIL_53] (rows=36524 width=1123)
                              predicate:(d_date BETWEEN '2000-01-27' AND '2000-04-27' and d_date_sk is not null)
                              TableScan [TS_3] (rows=73049 width=1123)
                                tpcds_bin_partitioned_orc_10@date_dim,d,Tbl:COMPLETE,Col:NONE,Output:["d_date_sk","d_date"]
                        Dynamic Partitioning Event Operator [EVENT_59] (rows=36524 width=1123)
                          Group By Operator [GBY_58] (rows=36524 width=1123)
                            Output:["_col0"],keys:_col0
                            Select Operator [SEL_57] (rows=36524 width=1123)
                              Output:["_col0"]
                               Please refer to the previous Select Operator [SEL_5]
                  <-Reducer 9 [SIMPLE_EDGE]
                    SHUFFLE [RS_29]
                      PartitionCols:_col0
                      Select Operator [SEL_26] (rows=7881108 width=259)
                        Output:["_col0","_col1"]
                        Group By Operator [GBY_25] (rows=7881108 width=259)
                          Output:["_col0","_col1"],aggregations:["avg(VALUE._col0)"],keys:KEY._col0
                        <-Reducer 8 [SIMPLE_EDGE]
                          SHUFFLE [RS_24]
                            PartitionCols:_col0
                            Group By Operator [GBY_23] (rows=15762217 width=259)
                              Output:["_col0","_col1"],aggregations:["avg(_col1)"],keys:_col0
                              Merge Join Operator [MERGEJOIN_64] (rows=15762217 width=259)
                                Conds:RS_19._col2=RS_20._col0(Inner),Output:["_col0","_col1"]
                              <-Map 10 [SIMPLE_EDGE]
                                SHUFFLE [RS_20]
                                  PartitionCols:_col0
                                  Select Operator [SEL_18] (rows=36524 width=1123)
                                    Output:["_col0"]
                                    Filter Operator [FIL_56] (rows=36524 width=1123)
                                      predicate:(d_date BETWEEN '2000-01-27' AND '2000-04-27' and d_date_sk is not null)
                                      TableScan [TS_16] (rows=73049 width=1123)
                                        tpcds_bin_partitioned_orc_10@date_dim,d,Tbl:COMPLETE,Col:NONE,Output:["d_date_sk","d_date"]
                                Dynamic Partitioning Event Operator [EVENT_62] (rows=36524 width=1123)
                                  Group By Operator [GBY_61] (rows=36524 width=1123)
                                    Output:["_col0"],keys:_col0
                                    Select Operator [SEL_60] (rows=36524 width=1123)
                                      Output:["_col0"]
                                       Please refer to the previous Select Operator [SEL_18]
                              <-Map 7 [SIMPLE_EDGE]
                                SHUFFLE [RS_19]
                                  PartitionCols:_col2
                                  Select Operator [SEL_15] (rows=14329288 width=259)
                                    Output:["_col0","_col1","_col2"]
                                    Filter Operator [FIL_55] (rows=14329288 width=259)
                                      predicate:cs_item_sk is not null
                                      TableScan [TS_13] (rows=14329288 width=259)
                                        tpcds_bin_partitioned_orc_10@catalog_sales,cs,Tbl:COMPLETE,Col:NONE,Output:["cs_item_sk","cs_ext_discount_amt"]