Uploaded image for project: 'IMPALA'
  1. IMPALA
  2. IMPALA-3760

Planner creates in-efficient join type for TPC-DS Q78

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Open
    • Priority: Minor
    • Resolution: Unresolved
    • Affects Version/s: Impala 2.5.0
    • Fix Version/s: None
    • Component/s: Frontend

      Description

      19:HASH JOIN should be a PARTITIONED join yet the planner picks a broadcast join, choosing BROADCAST Vs. PARTITIONED results in a 6x slowdown in the query.

      The wrong decision might be due to incorrect cardinality estimation of build and probe side for join 19.

      Snippet for problematic join with BROADCAST

            HASH_JOIN_NODE (id=19):(Total: 8m54s, non-child: 4m5s, % non-child: 46.00%)
               - BuildPartitionTime: 1m41s
               - BuildRows: 771.05M (771053773)
               - BuildRowsPartitioned: 1.54B (1541927827)
               - BuildTime: 2m9s
               - GetNewBlockTime: 6s148ms
               - HashBuckets: 2.15B (2147483648)
               - HashCollisions: 142.75K (142747)
               - LargestPartitionPercent: 6 (6)
               - MaxPartitionLevel: 1 (1)
               - NumRepartitions: 16 (16)
               - PartitionsCreated: 272 (272)
               - PeakMemoryUsage: 32.91 GB (35341572224)
               - PinTime: 11s094ms
               - ProbeRows: 3.22K (3222)
               - ProbeRowsPartitioned: 3.22K (3222)
               - ProbeTime: 23.812ms
               - RowsReturned: 84.11K (84111)
               - RowsReturnedRate: 156.00 /sec
               - SpilledPartitions: 16 (16)
               - UnpinTime: 11.937ms
      EXCHANGE_NODE (id=36):(Total: 4m14s, non-child: 51s153ms, % non-child: 20.14%)
                   - BytesReceived: 22.26 GB (23906000649)
                   - ConvertRowBatchTime: 11s254ms
                   - DeserializeRowBatchTimer: 57s224ms
                   - FirstBatchArrivalWaitTime: 46s181ms
                   - PeakMemoryUsage: 0
                   - RowsReturned: 771.05M (771053773)
                   - RowsReturnedRate: 3.04 M/sec
                   - SendersBlockedTimer: 997.004ms
                   - SendersBlockedTotalTimer(*): 23s133ms
      

      Snippet for fast PARTITIONED join

          HASH_JOIN_NODE (id=19):(Total: 1m21s, non-child: 320.967ms, % non-child: 0.40%)
               - BuildPartitionTime: 2s294ms
               - BuildRows: 12.64M (12640225)
               - BuildRowsPartitioned: 12.64M (12640225)
               - BuildTime: 1s989ms
               - GetNewBlockTime: 2.918ms
               - HashBuckets: 33.55M (33554432)
               - HashCollisions: 0 (0)
               - LargestPartitionPercent: 6 (6)
               - MaxPartitionLevel: 0 (0)
               - NumRepartitions: 0 (0)
               - PartitionsCreated: 16 (16)
               - PeakMemoryUsage: 1.66 GB (1782265446)
               - PinTime: 0.000ns
               - ProbeRows: 1.69K (1690)
               - ProbeRowsPartitioned: 0 (0)
               - ProbeTime: 9.016ms
               - RowsReturned: 44.12K (44124)
               - RowsReturnedRate: 543.00 /sec
               - SpilledPartitions: 0 (0)
               - UnpinTime: 0.000ns
      EXCHANGE_NODE (id=37):(Total: 55s735ms, non-child: 55s735ms, % non-child: 100.00%)
                 - BytesReceived: 390.28 MB (409241789)
                 - ConvertRowBatchTime: 401.360ms
                 - DeserializeRowBatchTimer: 1s451ms
                 - FirstBatchArrivalWaitTime: 45s286ms
                 - PeakMemoryUsage: 0
                 - RowsReturned: 12.64M (12640225)
                 - RowsReturnedRate: 226.81 K/sec
                 - SendersBlockedTimer: 14.032ms
                 - SendersBlockedTotalTimer(*): 713.074ms
      

      Query

      with ws as
        (select d_year AS ws_sold_year, ws_item_sk,
          ws_bill_customer_sk ws_customer_sk,
          sum(ws_quantity) ws_qty,
          sum(ws_wholesale_cost) ws_wc,
          sum(ws_sales_price) ws_sp
         from web_sales
         left join web_returns on wr_order_number=ws_order_number and ws_item_sk=wr_item_sk
         join date_dim on ws_sold_date_sk = d_date_sk
         where wr_order_number is null
         group by d_year, ws_item_sk, ws_bill_customer_sk
         ),
      cs as
        (select d_year AS cs_sold_year, cs_item_sk,
          cs_bill_customer_sk cs_customer_sk,
          sum(cs_quantity) cs_qty,
          sum(cs_wholesale_cost) cs_wc,
          sum(cs_sales_price) cs_sp
         from catalog_sales
         left join catalog_returns on cr_order_number=cs_order_number and cs_item_sk=cr_item_sk
         join date_dim on cs_sold_date_sk = d_date_sk
         where cr_order_number is null
         group by d_year, cs_item_sk, cs_bill_customer_sk
         ),
      ss as
        (select d_year AS ss_sold_year, ss_item_sk,
          ss_customer_sk,
          sum(ss_quantity) ss_qty,
          sum(ss_wholesale_cost) ss_wc,
          sum(ss_sales_price) ss_sp
         from store_sales
         left join store_returns on sr_ticket_number=ss_ticket_number and ss_item_sk=sr_item_sk
         join date_dim on ss_sold_date_sk = d_date_sk
         where sr_ticket_number is null
         group by d_year, ss_item_sk, ss_customer_sk
         )
      select 
      ss_sold_year, ss_item_sk, ss_customer_sk,
      round(ss_qty/(coalesce(ws_qty+cs_qty,1)),2) ratio,
      ss_qty store_qty, ss_wc store_wholesale_cost, ss_sp store_sales_price,
      coalesce(ws_qty,0)+coalesce(cs_qty,0) other_chan_qty,
      coalesce(ws_wc,0)+coalesce(cs_wc,0) other_chan_wholesale_cost,
      coalesce(ws_sp,0)+coalesce(cs_sp,0) other_chan_sales_price
      from ss
      left join ws on (ws_sold_year=ss_sold_year and ws_item_sk=ss_item_sk and ws_customer_sk=ss_customer_sk)
      left join cs on (cs_sold_year=ss_sold_year and cs_item_sk=cs_item_sk and cs_customer_sk=ss_customer_sk)
      where coalesce(ws_qty,0)>0 and coalesce(cs_qty, 0)>0 and ss_sold_year=2000
      order by 
        ss_sold_year, ss_item_sk, ss_customer_sk,
        store_qty desc, store_wholesale_cost desc, store_sales_price desc,
        other_chan_qty,
        other_chan_wholesale_cost,
        other_chan_sales_price,
        ratio
      limit 100
      

      Inefficient plan

      37:MERGING-EXCHANGE [UNPARTITIONED]
      |  order by: ss_sold_year ASC, ss_item_sk ASC, ss_customer_sk ASC, ss_qty DESC,
      |  limit: 100
      |  hosts=32 per-host-mem=unavailable
      |  tuple-ids=15 row-size=136B cardinality=100
      |
      20:TOP-N [LIMIT=100]
      |  order by: ss_sold_year ASC, ss_item_sk ASC, ss_customer_sk ASC, ss_qty DESC, 
      |  hosts=32 per-host-mem=13.28KB
      |  tuple-ids=15 row-size=136B cardinality=100
      |
      19:HASH JOIN [LEFT OUTER JOIN, BROADCAST]
      |  hash predicates: d_year = d_year, ss_customer_sk = cs_bill_customer_sk
      |  other predicates: coalesce(sum(cs_quantity), 0) > 0
      |  hosts=32 per-host-mem=5.03GB
      |  tuple-ids=3,8N,13N row-size=168B cardinality=1766829035
      |
      |--36:EXCHANGE [BROADCAST]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=13 row-size=56B cardinality=87718532
      |  |
      |  35:AGGREGATE [FINALIZE]
      |  |  output: sum:merge(cs_quantity), sum:merge(cs_wholesale_cost), sum:merge(cs_sales_price)
      |  |  group by: d_year, cs_item_sk, cs_bill_customer_sk
      |  |  having: coalesce(sum(cs_quantity), 0) > 0
      |  |  hosts=32 per-host-mem=5.03GB
      |  |  tuple-ids=13 row-size=56B cardinality=87718532
      |  |
      |  34:EXCHANGE [HASH(d_year,cs_item_sk,cs_bill_customer_sk)]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=13 row-size=56B cardinality=87718532
      |  |
      |  17:AGGREGATE [STREAMING]
      |  |  output: sum(cs_quantity), sum(cs_wholesale_cost), sum(cs_sales_price)
      |  |  group by: d_year, cs_item_sk, cs_bill_customer_sk
      |  |  hosts=32 per-host-mem=5.03GB
      |  |  tuple-ids=13 row-size=56B cardinality=87718532
      |  |
      |  16:HASH JOIN [INNER JOIN, BROADCAST]
      |  |  hash predicates: cs_sold_date_sk = d_date_sk
      |  |  runtime filters: RF002 <- d_date_sk
      |  |  hosts=32 per-host-mem=3.21KB
      |  |  tuple-ids=10,11N,12 row-size=60B cardinality=87718532
      |  |
      |  |--33:EXCHANGE [BROADCAST]
      |  |  |  hosts=1 per-host-mem=0B
      |  |  |  tuple-ids=12 row-size=8B cardinality=373
      |  |  |
      |  |  14:SCAN HDFS [date_dim, RANDOM]
      |  |     partitions=1/1 files=1 size=2.15MB
      |  |     predicates: date_dim.d_year = 2000
      |  |     table stats: 73049 rows total
      |  |     column stats: all
      |  |     hosts=1 per-host-mem=32.00MB
      |  |     tuple-ids=12 row-size=8B cardinality=373
      |  |
      |  15:HASH JOIN [LEFT OUTER JOIN, PARTITIONED]
      |  |  hash predicates: cs_order_number = cr_order_number, cs_item_sk = cr_item_sk
      |  |  other predicates: cr_order_number IS NULL
      |  |  hosts=32 per-host-mem=22.66MB
      |  |  tuple-ids=10,11N row-size=52B cardinality=432007888
      |  |
      |  |--32:EXCHANGE [HASH(cr_order_number,cr_item_sk)]
      |  |  |  hosts=32 per-host-mem=0B
      |  |  |  tuple-ids=11 row-size=16B cardinality=43201803
      |  |  |
      |  |  13:SCAN HDFS [catalog_returns, RANDOM]
      |  |     partitions=2104/2104 files=2104 size=32.99GB
      |  |     predicates: catalog_returns.cr_item_sk = catalog_returns.cr_item_sk
      |  |     table stats: 432018033 rows total
      |  |     column stats: all
      |  |     hosts=32 per-host-mem=64.00MB
      |  |     tuple-ids=11 row-size=16B cardinality=43201803
      |  |
      |  31:EXCHANGE [HASH(cs_order_number,cs_item_sk)]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=10 row-size=36B cardinality=432007888
      |  |
      |  12:SCAN HDFS [catalog_sales, RANDOM]
      |     partitions=1837/1837 files=2143 size=282.31GB
      |     predicates: catalog_sales.cs_item_sk = catalog_sales.cs_item_sk
      |     runtime filters: RF002 -> cs_sold_date_sk
      |     table stats: 4320078880 rows total
      |     column stats: all
      |     hosts=32 per-host-mem=528.00MB
      |     tuple-ids=10 row-size=36B cardinality=432007888
      |
      18:HASH JOIN [LEFT OUTER JOIN, PARTITIONED]
      |  hash predicates: d_year = d_year, ss_item_sk = ws_item_sk, ss_customer_sk = ws_bill_customer_sk
      |  other predicates: coalesce(sum(ws_quantity), 0) > 0
      |  hosts=32 per-host-mem=810.89MB
      |  tuple-ids=3,8N row-size=112B cardinality=1766829035
      |
      |--30:AGGREGATE [FINALIZE]
      |  |  output: sum:merge(ws_quantity), sum:merge(ws_wholesale_cost), sum:merge(ws_sales_price)
      |  |  group by: d_year, ws_item_sk, ws_bill_customer_sk
      |  |  having: coalesce(sum(ws_quantity), 0) > 0
      |  |  hosts=32 per-host-mem=25.34GB
      |  |  tuple-ids=8 row-size=56B cardinality=441704163
      |  |
      |  29:EXCHANGE [HASH(d_year,ws_item_sk,ws_bill_customer_sk)]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=8 row-size=56B cardinality=441704163
      |  |
      |  11:AGGREGATE [STREAMING]
      |  |  output: sum(ws_quantity), sum(ws_wholesale_cost), sum(ws_sales_price)
      |  |  group by: d_year, ws_item_sk, ws_bill_customer_sk
      |  |  hosts=32 per-host-mem=25.34GB
      |  |  tuple-ids=8 row-size=56B cardinality=441704163
      |  |
      |  10:HASH JOIN [INNER JOIN, BROADCAST]
      |  |  hash predicates: ws_sold_date_sk = d_date_sk
      |  |  runtime filters: RF001 <- d_date_sk
      |  |  hosts=32 per-host-mem=3.21KB
      |  |  tuple-ids=5,6N,7 row-size=60B cardinality=441704163
      |  |
      |  |--28:EXCHANGE [BROADCAST]
      |  |  |  hosts=1 per-host-mem=0B
      |  |  |  tuple-ids=7 row-size=8B cardinality=373
      |  |  |
      |  |  08:SCAN HDFS [date_dim, RANDOM]
      |  |     partitions=1/1 files=1 size=2.15MB
      |  |     predicates: date_dim.d_year = 2000
      |  |     table stats: 73049 rows total
      |  |     column stats: all
      |  |     hosts=1 per-host-mem=32.00MB
      |  |     tuple-ids=7 row-size=8B cardinality=373
      |  |
      |  09:HASH JOIN [LEFT OUTER JOIN, PARTITIONED]
      |  |  hash predicates: ws_order_number = wr_order_number, ws_item_sk = wr_item_sk
      |  |  other predicates: wr_order_number IS NULL
      |  |  hosts=32 per-host-mem=113.30MB
      |  |  tuple-ids=5,6N row-size=52B cardinality=2159968881
      |  |
      |  |--27:EXCHANGE [HASH(wr_order_number,wr_item_sk)]
      |  |  |  hosts=32 per-host-mem=0B
      |  |  |  tuple-ids=6 row-size=16B cardinality=216003761
      |  |  |
      |  |  07:SCAN HDFS [web_returns, RANDOM]
      |  |     partitions=2185/2185 files=2186 size=16.89GB
      |  |     table stats: 216003761 rows total
      |  |     column stats: all
      |  |     hosts=32 per-host-mem=32.00MB
      |  |     tuple-ids=6 row-size=16B cardinality=216003761
      |  |
      |  26:EXCHANGE [HASH(ws_order_number,ws_item_sk)]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=5 row-size=36B cardinality=2159968881
      |  |
      |  06:SCAN HDFS [web_sales, RANDOM]
      |     partitions=1824/1824 files=1824 size=138.28GB
      |     runtime filters: RF001 -> ws_sold_date_sk
      |     table stats: 2159968881 rows total
      |     column stats: all
      |     hosts=32 per-host-mem=528.00MB
      |     tuple-ids=5 row-size=36B cardinality=2159968881
      |
      25:AGGREGATE [FINALIZE]
      |  output: sum:merge(ss_quantity), sum:merge(ss_wholesale_cost), sum:merge(ss_sales_price)
      |  group by: d_year, ss_item_sk, ss_customer_sk
      |  hosts=32 per-host-mem=101.36GB
      |  tuple-ids=3 row-size=56B cardinality=1766829035
      |
      24:EXCHANGE [HASH(d_year,ss_item_sk,ss_customer_sk)]
      |  hosts=32 per-host-mem=0B
      |  tuple-ids=3 row-size=56B cardinality=1766829035
      |
      05:AGGREGATE [STREAMING]
      |  output: sum(ss_quantity), sum(ss_wholesale_cost), sum(ss_sales_price)
      |  group by: d_year, ss_item_sk, ss_customer_sk
      |  hosts=32 per-host-mem=101.36GB
      |  tuple-ids=3 row-size=56B cardinality=1766829035
      |
      04:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: ss_sold_date_sk = d_date_sk
      |  runtime filters: RF000 <- d_date_sk
      |  hosts=32 per-host-mem=3.21KB
      |  tuple-ids=0,1N,2 row-size=60B cardinality=1766829035
      |
      |--23:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=2 row-size=8B cardinality=373
      |  |
      |  02:SCAN HDFS [date_dim, RANDOM]
      |     partitions=1/1 files=1 size=2.15MB
      |     predicates: date_dim.d_year = 2000
      |     table stats: 73049 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=32.00MB
      |     tuple-ids=2 row-size=8B cardinality=373
      |
      03:HASH JOIN [LEFT OUTER JOIN, PARTITIONED]
      |  hash predicates: ss_ticket_number = sr_ticket_number, ss_item_sk = sr_item_sk
      |  other predicates: sr_ticket_number IS NULL
      |  hosts=32 per-host-mem=453.18MB
      |  tuple-ids=0,1N row-size=52B cardinality=8639936081
      |
      |--22:EXCHANGE [HASH(sr_ticket_number,sr_item_sk)]
      |  |  hosts=32 per-host-mem=0B
      |  |  tuple-ids=1 row-size=16B cardinality=863989652
      |  |
      |  01:SCAN HDFS [store_returns, RANDOM]
      |     partitions=2004/2004 files=2008 size=48.75GB
      |     table stats: 863989652 rows total
      |     column stats: all
      |     hosts=32 per-host-mem=80.00MB
      |     tuple-ids=1 row-size=16B cardinality=863989652
      |
      21:EXCHANGE [HASH(ss_ticket_number,ss_item_sk)]
      |  hosts=32 per-host-mem=0B
      |  tuple-ids=0 row-size=36B cardinality=8639936081
      |
      00:SCAN HDFS [store_sales, RANDOM]
         partitions=1824/1824 files=2649 size=376.63GB
         runtime filters: RF000 -> ss_sold_date_sk
         table stats: 8639936081 rows total
         column stats: all
         hosts=32 per-host-mem=528.00MB
         tuple-ids=0 row-size=36B cardinality=8639936081
      

      Query

      with ws as
        (select d_year AS ws_sold_year, ws_item_sk,
          ws_bill_customer_sk ws_customer_sk,
          sum(ws_quantity) ws_qty,
          sum(ws_wholesale_cost) ws_wc,
          sum(ws_sales_price) ws_sp
         from web_sales
         left join web_returns on wr_order_number=ws_order_number and ws_item_sk=wr_item_sk
         join date_dim on ws_sold_date_sk = d_date_sk
         where wr_order_number is null
         group by d_year, ws_item_sk, ws_bill_customer_sk
         ),
      cs as
        (select d_year AS cs_sold_year, cs_item_sk,
          cs_bill_customer_sk cs_customer_sk,
          sum(cs_quantity) cs_qty,
          sum(cs_wholesale_cost) cs_wc,
          sum(cs_sales_price) cs_sp
         from catalog_sales
         left join catalog_returns on cr_order_number=cs_order_number and cs_item_sk=cr_item_sk
         join date_dim on cs_sold_date_sk = d_date_sk
         where cr_order_number is null
         group by d_year, cs_item_sk, cs_bill_customer_sk
         ),
      ss as
        (select d_year AS ss_sold_year, ss_item_sk,
          ss_customer_sk,
          sum(ss_quantity) ss_qty,
          sum(ss_wholesale_cost) ss_wc,
          sum(ss_sales_price) ss_sp
         from store_sales
         left join store_returns on sr_ticket_number=ss_ticket_number and ss_item_sk=sr_item_sk
         join date_dim on ss_sold_date_sk = d_date_sk
         where sr_ticket_number is null
         group by d_year, ss_item_sk, ss_customer_sk
         )
      select 
      ss_sold_year, ss_item_sk, ss_customer_sk,
      round(ss_qty/(coalesce(ws_qty+cs_qty,1)),2) ratio,
      ss_qty store_qty, ss_wc store_wholesale_cost, ss_sp store_sales_price,
      coalesce(ws_qty,0)+coalesce(cs_qty,0) other_chan_qty,
      coalesce(ws_wc,0)+coalesce(cs_wc,0) other_chan_wholesale_cost,
      coalesce(ws_sp,0)+coalesce(cs_sp,0) other_chan_sales_price
      from ss
      left join ws on (ws_sold_year=ss_sold_year and ws_item_sk=ss_item_sk and ws_customer_sk=ss_customer_sk)
      left join cs on (cs_sold_year=ss_sold_year and cs_item_sk=cs_item_sk and cs_customer_sk=ss_customer_sk)
      where coalesce(ws_qty,0)>0 and coalesce(cs_qty, 0)>0 and ss_sold_year=2000
      order by 
        ss_sold_year, ss_item_sk, ss_customer_sk,
        store_qty desc, store_wholesale_cost desc, store_sales_price desc,
        other_chan_qty,
        other_chan_wholesale_cost,
        other_chan_sales_price,
        ratio
      limit 100
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                Unassigned
                Reporter:
                mmokhtar Mostafa Mokhtar
              • Votes:
                0 Vote for this issue
                Watchers:
                4 Start watching this issue

                Dates

                • Created:
                  Updated: