[
https://issues.apache.org/jira/browse/IMPALA-3760?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Greg Rahn updated IMPALA-3760:
------------------------------
Labels: performance tpc-ds (was: performance)
> Planner creates in-efficient join type for TPC-DS Q78
> -----------------------------------------------------
>
> Key: IMPALA-3760
> URL: https://issues.apache.org/jira/browse/IMPALA-3760
> Project: IMPALA
> Issue Type: Bug
> Components: Frontend
> Affects Versions: Impala 2.5.0
> Reporter: Mostafa Mokhtar
> Priority: Minor
> Labels: performance, tpc-ds
>
> 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
> {code}
> 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
> {code}
> Snippet for fast PARTITIONED join
> {code}
> 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
> {code}
> Query
> {code}
> 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
> {code}
> Inefficient plan
> {code}
> 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
> {code}
> Query
> {code}
> 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
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
