Stamatis Zampetakis created IMPALA-15489:
--------------------------------------------
Summary: Inefficient CTE plans with trivial IS NOT NULL filters
Key: IMPALA-15489
URL: https://issues.apache.org/jira/browse/IMPALA-15489
Project: IMPALA
Issue Type: Sub-task
Components: Frontend
Reporter: Stamatis Zampetakis
Assignee: Stamatis Zampetakis
A recent [small scale TPC-DS performance
run|https://jenkins.impala.io/job/perf-AB-test-ub2204/46/artifact/Impala/perf_results/latest/performance_result.txt]
(for IMPALA-15334) revealed some performance regressions on some queries.
{noformat}
+-----------+---------------------------+-----------------------+--------+-------------+------------+------------+----------------+-------+----------------+---------+---------+
| Workload | Query | File Format | Avg(s) | Base
Avg(s) | Delta(Avg) | StdDev(%) | Base StdDev(%) | Iters | Median Diff(%) | MW
Zval | Tval |
+-----------+---------------------------+-----------------------+--------+-------------+------------+------------+----------------+-------+----------------+---------+---------+
| TPCDS(20) | TPCDS-Q80A | parquet / none / none | 2.54 | 1.66
| R +53.15% | 0.74% | 0.93% | 25 | R +53.03% |
5.94 | 180.95 |
| TPCDS(20) | TPCDS-Q2 | parquet / none / none | 0.49 | 0.39
| R +24.68% | 6.04% | 8.49% | 25 | R +26.12% |
5.55 | 10.88 |
| TPCDS(20) | TPCDS-Q64 | parquet / none / none | 2.34 | 2.13
| R +9.76% | 0.80% | 0.98% | 25 | R +9.68% |
5.94 | 37.00 |
| TPCDS(20) | TPCDS-Q14-2 | parquet / none / none | 2.45 | 2.30
| R +6.33% | 1.25% | 0.61% | 25 | R +6.46% |
5.94 | 21.60 |
{noformat}
Even though IMPALA-15334 enabled more reuse the presence of certain CTEs have a
negative impact on performance.
The Q80A plan contains the following CTEs:
{noformat}
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_9])
ImpalaProjectRel(i_item_sk=[$0])
ImpalaFilterRel(condition=[AND(>($1, 50.00), IS NOT NULL($0))])
ImpalaProjectRel(i_item_sk=[$0], i_current_price=[$5])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, item]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_7])
ImpalaProjectRel(d_date_sk=[$0])
ImpalaFilterRel(condition=[AND(SEARCH(CAST($1):TIMESTAMP(15),
Sarg[[2000-08-23 00:00:00:TIMESTAMP(15)..2000-09-22
00:00:00:TIMESTAMP(15)]]:TIMESTAMP(15)), IS NOT NULL($0))])
ImpalaProjectRel(d_date_sk=[$0], d_date=[$2])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, date_dim]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_4])
ImpalaFilterRel(condition=[IS NOT NULL($0)])
ImpalaProjectRel(s_store_sk=[$0], s_store_id=[$1])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, store]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_5])
ImpalaFilterRel(condition=[IS NOT NULL($0)])
ImpalaProjectRel(web_site_sk=[$0], web_site_id=[$1])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, web_site]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_11])
ImpalaProjectRel(p_promo_sk=[$0])
ImpalaFilterRel(condition=[AND(=($1, _UTF-8'N'), IS NOT NULL($0))])
ImpalaProjectRel(p_promo_sk=[$0], p_channel_tv=[$11])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, promotion]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_2])
ImpalaFilterRel(condition=[AND(IS NOT NULL($6), IS NOT NULL($1), IS NOT
NULL($0), IS NOT NULL($2))])
ImpalaProjectRel(ws_item_sk=[$2], ws_web_site_sk=[$12], ws_promo_sk=[$15],
ws_order_number=[$16], ws_ext_sales_price=[$22], ws_net_profit=[$32],
ws_sold_date_sk=[$33])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, web_sales]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_3])
ImpalaFilterRel(condition=[IS NOT NULL($0)])
ImpalaProjectRel(cp_catalog_page_sk=[$0], cp_catalog_page_id=[$1])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, catalog_page]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_0])
ImpalaFilterRel(condition=[AND(IS NOT NULL($6), IS NOT NULL($0), IS NOT
NULL($1), IS NOT NULL($2))])
ImpalaProjectRel(cs_catalog_page_sk=[$11], cs_item_sk=[$14],
cs_promo_sk=[$15], cs_order_number=[$16], cs_ext_sales_price=[$22],
cs_net_profit=[$32], cs_sold_date_sk=[$33])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, catalog_sales]])
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_1])
ImpalaFilterRel(condition=[AND(IS NOT NULL($6), IS NOT NULL($1), IS NOT
NULL($0), IS NOT NULL($2))])
ImpalaProjectRel(ss_item_sk=[$1], ss_store_sk=[$6], ss_promo_sk=[$7],
ss_ticket_number=[$8], ss_ext_sales_price=[$14], ss_net_profit=[$21],
ss_sold_date_sk=[$22])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, store_sales]])
{noformat}
Observe that 7 out of the 9 CTEs in use are somewhat trivial and comprise only
{{NOT NULL}} filters over a table scan.
Let's focus on {{cte_suggestion_0}} which contains a bunch of {{NOT NULL}}
predicates over the big "catalog_sales" fact table.
{noformat}
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_0])
ImpalaFilterRel(condition=[AND(IS NOT NULL($6), IS NOT NULL($0), IS NOT
NULL($1), IS NOT NULL($2))])
ImpalaProjectRel(cs_catalog_page_sk=[$11], cs_item_sk=[$14],
cs_promo_sk=[$15], cs_order_number=[$16], cs_ext_sales_price=[$22],
cs_net_profit=[$32], cs_sold_date_sk=[$33])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, catalog_sales]])
{noformat}
>From the "ExecSummary" we can see that the plan with the _inefficient_ CTE on
>"catalog_sales" table is run once but is generally slower than plan without
>the CTE performing three individual scans.
*With CTE on catalog_sales*
{noformat}
| 15:CTE PRODUCER 3 12 123.901ms 207.259ms 28.55M
28.54M 143.94 MB 81.65 MB cte_suggestion_0, CANCELLED
| 16:SCAN HDFS 3 12 850.637ms 1s150ms 28.55M
28.54M 57.44 MB 88.00 MB tpcds20_parquet.catalog_sales
{noformat}
*Without CTE on catalog_sales*
{noformat}
| | 103:SCAN HDFS 3 12 180.306ms 652.859ms 35.63K
2.97M 57.31 MB 88.00 MB tpcds20_parquet.catalog_sales
| | 62:SCAN HDFS 3 12 137.640ms 486.274ms 35.63K
2.97M 57.31 MB 88.00 MB tpcds20_parquet.catalog_sales
| 23:SCAN HDFS 3 12 265.452ms 609.918ms 35.63K
2.97M 57.31 MB 88.00 MB tpcds20_parquet.catalog_sales
{noformat}
The number of rows emitted from the CTE scan is 1000x more that what is emitted
wihout it (28.55M vs. 35.63K). The difference in rows (and subsequently
performance) comes from runtime filters that cannot be applied when the CTE is
present.
{noformat}
16:SCAN HDFS [tpcds20_parquet.catalog_sales, RANDOM]
HDFS partitions=1/1 files=12 size=1.80GB
predicates: tpcds20_parquet.catalog_sales.cs_promo_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_sold_date_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_catalog_page_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_item_sk IS NOT NULL
stored statistics:
table: rows=28.80M size=1.80GB
| 23:SCAN HDFS [tpcds20_parquet.catalog_sales, RANDOM]
| HDFS partitions=1/1 files=12 size=1.80GB
| predicates: tpcds20_parquet.catalog_sales.cs_promo_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_sold_date_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_catalog_page_sk IS NOT NULL,
tpcds20_parquet.catalog_sales.cs_item_sk IS NOT NULL
| runtime filters: RF008[bloom] ->
tpcds20_parquet.catalog_sales.cs_catalog_page_sk, RF010[bloom] ->
tpcds20_parquet.catalog_sales.cs_promo_sk, RF012[bloom] ->
tpcds20_parquet.catalog_sales.cs_sold_date_sk, RF014[bloom] ->
tpcds20_parquet.catalog_sales.cs_item_sk
| stored statistics:
| table: rows=28.80M size=1.80GB{noformat}
Another important observation is that the number of rows emitted from the CTE
is very close to the actual rows of the table (28.55M vs 28.80M) so the
multiple {{NOT NULL}} predicates are low selective.
Low selective {{NOT_NULL}} filters over small dimension tables, such as
"catalog_page", do not help either in runtime performance althouth the results
are not as detrimental as in fact tables.
{noformat}
ImpalaCTEProducer(readType=[EAGER], writeType=[EAGER],
cteName=[cte_suggestion_3])
ImpalaFilterRel(condition=[IS NOT NULL($0)])
ImpalaProjectRel(cp_catalog_page_sk=[$0], cp_catalog_page_id=[$1])
ImpalaHdfsScanRel(table=[[tpcds_partitioned_parquet_snap, catalog_page]])
{noformat}
*With CTE on catalog_page*
{noformat}
| | 13:CTE PRODUCER 1 1 20.135us 20.135us 11.72K
11.72K 6.53 MB 6.00 MB cte_suggestion_3, CANCELLED
| | 14:SCAN HDFS 1 1 29.265ms 29.265ms 11.72K
11.72K 745.08 KB 16.00 MB tpcds20_parquet.catalog_page
{noformat}
*Without CTE on catalog_page*
{noformat}
| | | F63:EXCHANGE SENDER 1 1 6.143ms 6.143ms
198.15 KB 144.00 KB
| | | 112:SCAN HDFS 1 1 7.107ms 7.107ms 11.72K
11.72K 745.08 KB 16.00 MB tpcds20_parquet.catalog_page
| | | F36:EXCHANGE SENDER 1 1 6.528ms 6.528ms
198.15 KB 144.00 KB
| | | 71:SCAN HDFS 1 1 2.709ms 2.709ms 11.72K
11.72K 745.08 KB 16.00 MB tpcds20_parquet.catalog_page
| | F12:EXCHANGE SENDER 1 1 1.094ms 1.094ms
198.15 KB 144.00 KB
| | 32:SCAN HDFS 1 1 9.790ms 9.790ms 11.72K
11.72K 745.08 KB 16.00 MB tpcds20_parquet.catalog_page
{noformat}
Overall, choosing to use a CTE for low selective {{NOT NULL}} filters is
inefficient and this is a very frequent pattern in the TPC-DS benchmark and
especially prominent in Q80A.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
