[IMPALA-3760] Planner creates in-efficient join type for TPC-DS Q78 - ASF JIRA

XML

Word

Printable

JSON

Details

Type: Bug
Status: Open
Priority: Minor
Resolution: Unresolved
Affects Version/s: Impala 2.5.0
Fix Version/s: None
Component/s: Frontend
Labels:
- performance
- tpc-ds

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

relates to

IMPALA-2932 Extend DistributedPlanner::createHashJoinFragment to account for hash table build cost

Resolved

Activity

People

Assignee:: Unassigned

Reporter:: Mostafa Mokhtar

Votes:: 0 Vote for this issue

Watchers:: 4 Start watching this issue

Dates

Created:: 20/Jun/16 06:11

Updated:: 07/Mar/19 06:25