[
https://issues.apache.org/jira/browse/IMPALA-10681?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17351552#comment-17351552
]
ASF subversion and git services commented on IMPALA-10681:
----------------------------------------------------------
Commit 954eb5c85d329af7690698cdc4d0f409260e6d18 in impala's branch
refs/heads/master from Aman Sinha
[ https://gitbox.apache.org/repos/asf?p=impala.git;h=954eb5c ]
IMPALA-10681: Improve inner join cardinality estimates
During cardinality estimation for inner joins, if the join
conjunct involves a scan slot on left side and a function
(e.g MAX) on the right, currently we determine that the NDV
stats of either side is not useful and return the left side's
cardinality even though it may be a significant over-estimate.
In this patch, we handle join conjuncts of such types by
keeping them in an 'other' eligible conjuncts list as long as
the NDV for expressions on both sides of the join and the
input row count is available. For example, in the following
cases the NDV is available but was not being used for inner
joins since the previous logic was only looking for scan
slots: (a) int_col = MAX(int_col) and the right input does
not have a group-by, so right NDV = 1 can be used. (b) if it
has a group-by and the group-by columns already have
associated NDV, the combined NDV is also available.
Other such examples exist. An auxiliary struct is introduced
to keep track of the ndv and row count.
Once these 'other' eligible conjuncts are populated, we do the
join cardinality estimation in a manner similar to the normal
join conjuncts by fetching the stats from the auxiliary struct.
Testing:
- Added new planner tests for inner join cardinality
- Modified expected plans for certains tests including
TPC-DS queries and ran end-to-end TPC-DS queries
- Since TPC-DS plans are complex, I did a check of the cardinality
changes for some of the hash joins but not the changes in the
shape of a plan (e.g whether the join order changed).
- Preliminary tests with a TPC-DS 10 GB scale factor on a single
node showed between 5-15% performance improvements for 4 of the
6 queries whose plans changed.
Change-Id: I8aa9d3b8f3c4848b3e9414fe19ad7ad348d12ecc
Reviewed-on: http://gerrit.cloudera.org:8080/17387
Reviewed-by: Impala Public Jenkins <[email protected]>
Tested-by: Aman Sinha <[email protected]>
Reviewed-by: Aman Sinha <[email protected]>
> JOIN cardinality is wrong for INNER joins when combined with aggregations
> -------------------------------------------------------------------------
>
> Key: IMPALA-10681
> URL: https://issues.apache.org/jira/browse/IMPALA-10681
> Project: IMPALA
> Issue Type: Bug
> Components: Frontend
> Reporter: Zoltán Borók-Nagy
> Assignee: Aman Sinha
> Priority: Major
>
> JOIN cardinality estimate can be off for INNER joins. Consider the following
> LEFT SEMI JOIN which estimates the cardinalities well:
> {noformat}
> [localhost:21050] tpcds_parquet> explain select * from store_sales left semi
> join (select max(s_store_sk) as max_store_sk from store) v on ss_store_sk =
> max_store_sk;
> Query: explain select * from store_sales left semi join (select
> max(s_store_sk) as max_store_sk from store) v on ss_store_sk = max_store_sk
> +-------------------------------------------------------------+
> | Explain String |
> +-------------------------------------------------------------+
> | Max Per-Host Resource Reservation: Memory=14.95MB Threads=6 |
> | Per-Host Resource Estimates: Memory=139MB |
> | |
> | PLAN-ROOT SINK |
> | | |
> | 07:EXCHANGE [UNPARTITIONED] |
> | | |
> | 03:HASH JOIN [LEFT SEMI JOIN, BROADCAST] |
> | | hash predicates: ss_store_sk = max(s_store_sk) |
> | | runtime filters: RF000 <- max(s_store_sk) |
> | | row-size=100B cardinality=480.07K |
> | | |
> | |--06:EXCHANGE [BROADCAST] |
> | | | |
> | | 05:AGGREGATE [FINALIZE] |
> | | | output: max:merge(s_store_sk) |
> | | | row-size=4B cardinality=1 |
> | | | |
> | | 04:EXCHANGE [UNPARTITIONED] |
> | | | |
> | | 02:AGGREGATE |
> | | | output: max(s_store_sk) |
> | | | row-size=4B cardinality=1 |
> | | | |
> | | 01:SCAN HDFS [tpcds_parquet.store] |
> | | HDFS partitions=1/1 files=1 size=9.93KB |
> | | row-size=4B cardinality=12 |
> | | |
> | 00:SCAN HDFS [tpcds_parquet.store_sales] |
> | HDFS partitions=1824/1824 files=1824 size=200.93MB |
> | runtime filters: RF000 -> ss_store_sk |
> | row-size=100B cardinality=2.88M |
> +-------------------------------------------------------------+
> {noformat}
> JOIN cardinality is 1/6 of LHS scan node cardinality which seems reasonable,
> since LHS NDV is 6, and the right side only has one row.
> Now let's switch to an INNER join:
> {noformat}
> [localhost:21050] tpcds_parquet> explain select * from store_sales inner join
> (select max(s_store_sk) as max_store_sk from store) v on ss_store_sk =
> max_store_sk;
> Query: explain select * from store_sales inner join (select max(s_store_sk)
> as max_store_sk from store) v on ss_store_sk = max_store_sk
> +-------------------------------------------------------------+
> | Explain String |
> +-------------------------------------------------------------+
> | Max Per-Host Resource Reservation: Memory=14.95MB Threads=6 |
> | Per-Host Resource Estimates: Memory=193MB |
> | |
> | PLAN-ROOT SINK |
> | | |
> | 07:EXCHANGE [UNPARTITIONED] |
> | | |
> | 03:HASH JOIN [INNER JOIN, BROADCAST] |
> | | hash predicates: ss_store_sk = max(s_store_sk) |
> | | runtime filters: RF000 <- max(s_store_sk) |
> | | row-size=104B cardinality=2.88M |
> | | |
> | |--06:EXCHANGE [BROADCAST] |
> | | | |
> | | 05:AGGREGATE [FINALIZE] |
> | | | output: max:merge(s_store_sk) |
> | | | row-size=4B cardinality=1 |
> | | | |
> | | 04:EXCHANGE [UNPARTITIONED] |
> | | | |
> | | 02:AGGREGATE |
> | | | output: max(s_store_sk) |
> | | | row-size=4B cardinality=1 |
> | | | |
> | | 01:SCAN HDFS [tpcds_parquet.store] |
> | | HDFS partitions=1/1 files=1 size=9.93KB |
> | | row-size=4B cardinality=12 |
> | | |
> | 00:SCAN HDFS [tpcds_parquet.store_sales] |
> | HDFS partitions=1824/1824 files=1824 size=200.93MB |
> | runtime filters: RF000 -> ss_store_sk |
> | row-size=100B cardinality=2.88M |
> +-------------------------------------------------------------+
> {noformat}
> The JOIN cardinality equals to the lhs cardinality even when the rhs
> cardinality is only one.
> SEMI JOIN cardinality is calculated differently than INNER join cardinality.
> SEMI JOIN cardinality:
> https://github.com/apache/impala/blob/c65d7861d9ae28f6fc592727ff699a8155dcda2c/fe/src/main/java/org/apache/impala/planner/JoinNode.java#L486-L562
> INNER JOIN cardinality:
> https://github.com/apache/impala/blob/c65d7861d9ae28f6fc592727ff699a8155dcda2c/fe/src/main/java/org/apache/impala/planner/JoinNode.java#L242-L308
> The problem is that the latter doesn't find the equi join conjunct
> "ss_store_sk = max(s_store_sk)" eligible, so it returns lhs cardinality:
> https://github.com/apache/impala/blob/c65d7861d9ae28f6fc592727ff699a8155dcda2c/fe/src/main/java/org/apache/impala/planner/JoinNode.java#L296-L300
> ss_store_sk = max(s_store_sk) is not eligible because Expr.findSrcScanSlot()
> returns NULL for "max(s_store_sk)."
> https://github.com/apache/impala/blob/c65d7861d9ae28f6fc592727ff699a8155dcda2c/fe/src/main/java/org/apache/impala/planner/JoinNode.java#L449
> I think the solution should be to either change Expr.findSrcScanSlot() to
> return the scan slot. Or, change getJoinCardinality() to return an estimation
> similar to the SEMI JOIN. Or fix both.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
